Protocol Module

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

Module implementing the Constellation communication protocols.

class core.protocol.MessageHeader(name: str, protocol: Protocol)

Bases: object

Class implementing a Constellation message header.

decode(header: Any) → Tuple[str, Timestamp, dict[str, Any] | None] | Tuple[str, int, int, dict[str, Any] | None]

Decode header string and return host, timestamp and meta map.

encode(meta: dict[str, Any] | None = None, **kwargs: Any) → memoryview

Generate and return a header as list.

Additional keyword arguments are required for protocols specifying additional fields.

recv(socket: Socket, flags: int = 0) → Tuple[str, Timestamp, dict[str, Any] | None] | Tuple[str, int, int, dict[str, Any] | None]

Receive header from socket and return all decoded fields.

send(socket: ~zmq.sugar.socket.Socket, flags: int = <Flag.SNDMORE: 2>, meta: dict[str, ~typing.Any] | None = None, **kwargs: ~typing.Any) → None

Send a message header via socket.

meta is an optional dictionary that is sent as a map of string/value pairs with the header.

class core.protocol.Protocol(*values)

Bases: StrEnum

CDTP1 = 'CDTP\x01'

CHP1 = 'CHP\x01'

CMDP1 = 'CMDP\x01'

CSCP1 = 'CSCP\x01'