Events Module

Event system for inter-component communication

The events module provides a centralized event system for communication between components.

Events Class

Manages event subscriptions, emissions, and scheduling.

Methods

enable()

def enable(self)

Enables the event system and starts the scheduler thread.

connect()

def connect(self, event_name, function)

Subscribes a callback function to an event.

event_name · str

Name of the event to subscribe to

function · callable

Callback function to execute when event is emitted

Example: Subscribe to Events

from pynicotine.events import events

def on_user_status(msg):
    print(f"User {msg.user} status: {msg.status}")

events.connect("user-status", on_user_status)

disconnect()

def disconnect(self, event_name, function)

Unsubscribes a callback function from an event.

event_name · str

Name of the event

function · callable

The callback function to remove

emit()

def emit(self, event_name, *args, **kwargs)

Emits an event synchronously to all subscribers.

event_name · str

Name of the event to emit

args · tuple

Positional arguments passed to callbacks

kwargs · dict

Keyword arguments passed to callbacks

Example: Emit Events

from pynicotine.events import events

events.emit("user-status", username="Alice", status=2)

emit_main_thread()

def emit_main_thread(self, event_name, *args, **kwargs)

Queues an event for emission in the main thread.

Use this method when emitting events from background threads to ensure thread safety.

schedule()

def schedule(self, delay, callback, callback_args=None, repeat=False)

Schedules a callback to run after a delay.

delay · float

Delay in seconds before executing callback

callback · callable

Function to execute

callback_args · tuple

Arguments to pass to callback

repeat · bool · default: false

Whether to repeat the callback periodically

event_id · int

Unique identifier for the scheduled event

Example: Schedule Callbacks

from pynicotine.events import events

def check_status():
    print("Checking status...")

# Run once after 5 seconds
event_id = events.schedule(5.0, check_status)

# Run every 10 seconds
events.schedule(10.0, check_status, repeat=True)

cancel_scheduled()

def cancel_scheduled(self, event_id)

Cancels a scheduled callback.

event_id · int

The event ID returned by schedule()

process_thread_events()

def process_thread_events(self)

Processes queued thread events in the main thread. Called by the main loop.

continue · bool

Whether the main loop should continue processing

Available Events

General Events

start · event

Application startup

quit · event

Application shutdown

setup · event

Initial setup/configuration screen

Server Events

server-login · event

Server login success or failure

server-disconnect · event

Disconnected from server

server-reconnect · event

Reconnecting to server

User Events

user-status · event

User status changed (online/away/offline)

user-stats · event

User statistics updated

user-country · event

User country code determined

Chat Events

say-chat-room · event

Message in chat room

message-user · event

Private message sent/received

join-room · event

Joined a chat room

leave-room · event

Left a chat room

Search Events

add-search · event

New search started

file-search-response · event

Search results received

Transfer Events

update-download · event

Download updated

update-upload · event

Upload updated

file-download-progress · event

Download progress update

file-upload-progress · event

Upload progress update

Example: Complete Event Usage

from pynicotine.events import events

class MyComponent:
    def __init__(self):
        # Subscribe to events
        events.connect("start", self._on_start)
        events.connect("user-status", self._on_user_status)
        events.connect("quit", self._on_quit)
        
        # Schedule periodic check
        self.timer_id = events.schedule(
            60.0, self._periodic_check, repeat=True
        )
    
    def _on_start(self):
        print("Application started")
    
    def _on_user_status(self, msg):
        print(f"User {msg.user}: {msg.status}")
    
    def _periodic_check(self):
        print("Periodic check")
    
    def _on_quit(self):
        # Clean up
        events.cancel_scheduled(self.timer_id)
        events.disconnect("start", self._on_start)
        events.disconnect("user-status", self._on_user_status)

Global Instance

from pynicotine.events import events

A global events instance is available for import.