BroadcastManager Module

SPDX-FileCopyrightText: 2024 DESY and the Constellation authors SPDX-License-Identifier: EUPL-1.2

BroadcastManger module provides classes for managing CHIRP broadcasts within Constellation Satellites.

class core.broadcastmanager.CHIRPBroadcaster(name: str, group: str, interface: list[str], mon_port: int | None = None, **kwds: Any)

Bases: BaseSatelliteFrame

Manages service discovery and broadcast via the CHIRP protocol.

Listening and reacting to CHIRP broadcasts is implemented in a dedicated thread that can be started after the class has been instantiated.

Discovered services are added to an internal cache. Callback methods can be registered either by calling register_request() or by using the @chirp_callback() decorator. The callback will be added to the satellite’s internal task queue once the corresponding service has been offered by other satellites via broadcast.

Offered services can be registered via register_offer() and are announced on incoming request broadcasts or via broadcast_offers().

broadcast_depart() → None

Broadcast DEPART for all registered services.

broadcast_offers(serviceid: CHIRPServiceIdentifier | None = None) → None

Broadcast all registered services matching serviceid.

Specify None for all registered services.

broadcast_requests() → None

Broadcast all requests registered via register_request().

get_discovered(serviceid: CHIRPServiceIdentifier) → list[DiscoveredService]

Return a list of already discovered services for a given identifier.

register_offer(serviceid: CHIRPServiceIdentifier, port: int) → None

Register new offered service or overwrite existing service.

register_request(serviceid: CHIRPServiceIdentifier, callback: Callable[[B, DiscoveredService], None]) → None

Register new callback for ServiceIdentifier.

request(serviceid: CHIRPServiceIdentifier) → None

Request specific service.

Should already have a registered callback via register_request(), or any incoming OFFERS will go unnoticed.

class core.broadcastmanager.DiscoveredService(host_uuid: UUID, serviceid: CHIRPServiceIdentifier, address: str, port: int, alive: bool = True)

Bases: object

Class to hold discovered service data.

core.broadcastmanager.chirp_callback(request_service: CHIRPServiceIdentifier) → Callable[[Callable[[P], T]], Callable[[P], T]]

Mark a function as a callback for CHIRP service requests.

core.broadcastmanager.get_chirp_callbacks(cls: object) → dict[CHIRPServiceIdentifier, Callable[[B, DiscoveredService], None]]

Loop over all class methods and return those marked as CHIRP callback.