tribler.core.restapi.events_endpoint

Attributes

topics_to_send_to_gui

Classes

`MessageDict`	A message fit for the GUI, usually forwarded from the Notifier.
`EventsEndpoint`	Important events in Tribler are returned over the events endpoint. This connection is held open. Each event is

Module Contents

tribler.core.restapi.events_endpoint.topics_to_send_to_gui

class tribler.core.restapi.events_endpoint.MessageDict

Bases: TypedDict

A message fit for the GUI, usually forwarded from the Notifier.

topic: str

kwargs: dict[str, str]

class tribler.core.restapi.events_endpoint.EventsEndpoint(notifier: tribler.core.notifier.Notifier, public_key: str | None = None)

Bases: tribler.core.restapi.rest_endpoint.RESTEndpoint

Important events in Tribler are returned over the events endpoint. This connection is held open. Each event is pushed over this endpoint in the form of a JSON dictionary. Each JSON dictionary contains a type field that indicates the type of the event. Individual events are separated by a newline character.

path = '/api/events'

shutdown_event

events_responses: list[aiohttp.web.StreamResponse] = []

undelivered_error: Exception | None = None

public_key = None

notifier

queue: asyncio.Queue[MessageDict]

property _shutdown: bool

on_notification(topic: tribler.core.notifier.Notification, **kwargs) → None: Callback for when a notification is received. Check if we should forward it to the GUI.

on_circuit_removed(circuit: ipv8.messaging.anonymization.tunnel.Circuit, additional_info: str) → None

Special handler for circuit removal notifications.

The original notification contains non-JSON-serializable argument, so we send another one to GUI.

initial_message() → MessageDict: Create the initial message to announce to the GUI.

error_message(reported_error: Exception) → MessageDict: Create an error message for the GUI.

encode_message(message: MessageDict) → bytes: Use JSON to dump the given message to bytes.

has_connection_to_gui() → bool: Whether the GUI has responded before.

should_skip_message(message: MessageDict) → bool: Returns True if EventsEndpoint should skip sending message to GUI due to a shutdown or no connection to GUI. Issue an appropriate warning if the message cannot be sent.

send_event(message: MessageDict) → None: Put event message to a queue to be sent to GUI.

async process_queue() → None: Get all failed messages in the queue and send them to the GUI.

async _write_data(message: MessageDict) → None: Write data over the event socket if it’s open.

on_tribler_exception(reported_error: Exception) → None: An exception has occurred in Tribler.

async get_info(request: aiohttp.web_request.Request) → tribler.core.restapi.rest_endpoint.RESTResponse: Get the general info of the events endpoint.

async get_events(request: aiohttp.web_request.Request) → aiohttp.web.StreamResponse

A GET request to this endpoint will open the event connection.

Example request:

curl -X GET http://localhost:20100/events