nvflare.fuel.f3.sfm.conn_manager module

class ConnManager(local_endpoint: Endpoint)

Bases: ConnMonitor

SFM connection manager The class is responsible for maintaining state of SFM connections and pumping data through them

add_connector(driver: Driver, params: dict, mode: Mode) → str

add_endpoint_monitor(monitor: EndpointMonitor)

close_connection(connection: Connection)

find_endpoint(name: str) → Endpoint | None

get_connections(name: str) → List[SfmConnection] | None

static get_dict_payload(prefix, frame)

handle_new_connection(connection: Connection)

loopback_message_task(endpoint: Endpoint, app_id: int, headers: dict | None, payload: bytes | bytearray | memoryview | list)

notify_monitors(endpoint: Endpoint)

process_frame(sfm_conn: SfmConnection, frame: bytes | bytearray | memoryview | list)

process_frame_task(sfm_conn: SfmConnection, frame: bytes | bytearray | memoryview | list)

register_message_receiver(app_id: int, receiver: MessageReceiver)

remove_connector(handle: str)

remove_endpoint(name: str)

send_loopback_message(endpoint: Endpoint, app_id: int, headers: dict | None, payload: bytes | bytearray | memoryview | list)

Send message to itself

send_message(endpoint: Endpoint, app_id: int, headers: dict | None, payload: bytes | bytearray | memoryview | list)

Send a message to endpoint for app

The message is asynchronous, no response is expected.

Parameters:

• endpoint – An endpoint to send the message to

• app_id – Application ID

• headers – headers, optional

• payload – message payload, optional

Raises:

CommError – If any error happens while sending the data

start()

start_connector(connector: ConnectorInfo)

Start connector in a new thread

static start_connector_task(connector: ConnectorInfo)

Start connector in a new thread This function will loop as long as connector is not stopped

state_change(connection: Connection)

Driver state change notification, including new connections

Parameters:

connection – The connection that state has changed

Raises:

CommError – If any error happens while processing the frame

stop()

update_endpoint(sfm_conn: SfmConnection, data: dict)

class NullConnection

Bases: Connection

A mock connection used for loopback messages

close()

Close connection

Raises:

CommError – If any errors

get_conn_properties() → dict

Get connection specific properties, like peer address, TLS certificate etc

Raises:

CommError – If any errors

send_frame(frame: bytes | bytearray | memoryview | list)

Send a SFM frame through the connection to the remote endpoint.

Parameters:

frame – The frame to be sent

Raises:

CommError – If any error happens while sending the frame

class SfmFrameReceiver(conn_manager: ConnManager, conn: SfmConnection)

Bases: FrameReceiver

process_frame(frame: bytes | bytearray | memoryview | list)

Frame received callback

Parameters:

frame – The frame received

Raises:

CommError – If any error happens while processing the frame

get_handle()