From 0c2ba31af1e93d0dd7f122e4b98a755604d5a1b0 Mon Sep 17 00:00:00 2001 From: Mike Langmayr Date: Mon, 2 Nov 2026 14:57:10 -0800 Subject: [PATCH 1/2] Add a Cyclone DDS transport carrying bamboo envelopes as typed DDS samples --- README.md | 14 +- docs/source/api/index.md | 6 + docs/source/cli.md | 15 +- docs/source/client.md | 4 +- docs/source/index.md | 2 +- docs/source/installation.md | 18 +- libby/cli/cli_config.example.yaml | 10 +- libby/cli/libby_cli.py | 28 ++- libby/client.py | 22 ++- libby/config_resolve.py | 19 ++- libby/cyclonedds_transport.py | 263 +++++++++++++++++++++++++++++ libby/daemon.py | 23 ++- libby/how_to_create_a_daemon.md | 40 ++++- libby/libby.py | 51 ++++++ pyproject.toml | 3 + tests/test_client_integration.py | 80 ++++++++- tests/test_cyclonedds_transport.py | 124 ++++++++++++++ tox.ini | 9 +- 18 files changed, 700 insertions(+), 31 deletions(-) create mode 100644 libby/cyclonedds_transport.py create mode 100644 tests/test_cyclonedds_transport.py diff --git a/README.md b/README.md index 932ad2e..5ad2b62 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Libby -Libby: a tiny messaging library which uses Bamboo with pluggable transports (ZMQ or RabbitMQ) +Libby: a tiny messaging library which uses Bamboo with pluggable transports (ZMQ, RabbitMQ or Cyclone DDS) ## Documentation @@ -37,21 +37,24 @@ library, the CLI, and writing a `LibbyDaemon` peer). `libby list .` (and `Client.peers`) is how you find out which peers are up. It broadcasts a `keys.list` that every peer answers, so it works -the same on both transports and needs nothing configured beyond the transport -itself. +the same on every transport and needs nothing configured beyond the transport +itself. Over Cyclone DDS the broadcast reaches every peer DDS discovery has +found. Discovery, in the sense of a peer table you can ask who is alive, is not implemented. It may be one day; until then use the broadcast above rather than these: - `Libby.peers_alive()` returns `{}` and `Libby.wait_for_peer()` always fails, - on either transport, because `Protocol` never instantiates a `PeerTable`. + on any transport, because `Protocol` never instantiates a `PeerTable`. - ZMQ runs bamboo's hello, but only toward peers already in the sender's address book, and nothing consumes it for liveness. A client learns nothing about a daemon that has not been told about the client, so `Libby.knows_key()` stays False in the usual one-sided setup. - `Libby.rabbitmq()` does not start hello at all. The broker routes messages; it does not tell a peer who else is connected. +- `Libby.cyclonedds()` does not start hello either. DDS discovery finds peers + for routing, but libby does not yet expose it as a peer table. `LibbyDaemon`'s `discovery_enabled` / `on_hello` control bamboo's hello, not the broadcast above. @@ -66,4 +69,5 @@ Most of `tests/` needs no transport at all. `tests/test_client_integration.py` is the exception: it starts a real `LibbyDaemon` and exercises `Client` against it, once per transport. The ZMQ cases need no external service and always run; the RabbitMQ cases skip themselves if no broker is reachable at -`amqp://localhost`. +`amqp://localhost`; the Cyclone DDS cases skip themselves unless the +`cyclonedds` extra is installed. diff --git a/docs/source/api/index.md b/docs/source/api/index.md index 7503ceb..bb85c26 100644 --- a/docs/source/api/index.md +++ b/docs/source/api/index.md @@ -133,6 +133,12 @@ the docs build does not install the optional `influxdb` extra. :show-inheritance: ``` +```{eval-rst} +.. automodule:: libby.cyclonedds_transport + :members: + :show-inheritance: +``` + ```{eval-rst} .. automodule:: libby._transport_errors :members: diff --git a/docs/source/cli.md b/docs/source/cli.md index 933e962..b305df7 100644 --- a/docs/source/cli.md +++ b/docs/source/cli.md @@ -15,7 +15,7 @@ libby completion bash|zsh # print shell code for TAB com `.` is the address of one daemon: `group` is its `group_id` and `daemon` is its `peer_id` (e.g. `peer_id: adc, group_id: hsfei` in the daemon's config is addressed as `hsfei.adc`). `Libby.rabbitmq()` / -`Libby.zmq()` build the actual wire identity from those two fields via +`Libby.zmq()` / `Libby.cyclonedds()` build the actual wire identity from those two fields via `libby.naming.qualified_peer_id`, so a daemon config never needs to concatenate them by hand. `group`/`daemon` are case-insensitive (`HSFEI.ADC` and `hsfei.adc` reach the same daemon); the keyword isn't. `list` is the one @@ -91,7 +91,9 @@ clients reply to the broadcast too, and they identify themselves as clients and are left out. Over ZMQ the broadcast only reaches daemons in the address book (`peers:` in -`cli_config.yaml`, or `--addr`); over RabbitMQ the broker reaches everyone. +`cli_config.yaml`, or `--addr`); over RabbitMQ the broker reaches everyone; +over Cyclone DDS it reaches every daemon DDS discovery has found on the +domain. ## Completion @@ -173,9 +175,13 @@ cp $(python -c "import libby.cli, os; print(os.path.dirname(libby.cli.__file__)) Schema: ```yaml -transport: rabbitmq # zmq | rabbitmq +transport: rabbitmq # zmq | rabbitmq | cyclonedds rabbitmq_url: amqp://localhost +# Used only when transport=cyclonedds: +dds_domain_id: 0 +dds_config: file:///etc/libby/cyclonedds.xml # optional; else CYCLONEDDS_URI + # Used only when transport=zmq: peers: hsfei_pickoff: tcp://hispec.caltech.edu:5555 @@ -184,7 +190,8 @@ peers: All keys are optional. Missing file is fine — defaults are `transport: rabbitmq` / `rabbitmq_url: amqp://localhost`. -Precedence: `--transport` / `--rabbitmq-url` flags override yaml; yaml +Precedence: `--transport` / `--rabbitmq-url` / `--dds-domain` / +`--dds-config` flags override yaml; yaml overrides built-in defaults. Flags must appear *after* the subcommand (`libby show --transport zmq foo`, not the other way). diff --git a/docs/source/client.md b/docs/source/client.md index 0159ae4..59a75cd 100644 --- a/docs/source/client.md +++ b/docs/source/client.md @@ -20,6 +20,7 @@ Construct explicitly when you don't want config-file resolution: ```python client = Client.rabbitmq(rabbitmq_url="amqp://user:pass@host") client = Client.zmq(address_book={"hsfei_pickoff": "tcp://host:5555"}) +client = Client.cyclonedds(domain_id=0) # needs the cyclonedds extra ``` - `get(name)` → the value; `show(name)` → the full response dict (value, @@ -132,7 +133,8 @@ These broadcast a single `keys.list`, which every daemon answers. Nothing says how many daemons exist, so they wait the full `timeout_s` (1s by default) rather than returning on the first reply, and a daemon that is down is simply absent. Over ZMQ the broadcast reaches only the daemons in the -address book; over RabbitMQ the broker reaches all of them. +address book; over RabbitMQ the broker reaches all of them; over Cyclone DDS +it reaches every daemon discovered on the domain. Only daemons come back. Every `Libby` serves `keys.list`, so another `Client` answers the broadcast as well, but `LibbyDaemon` is the only thing diff --git a/docs/source/index.md b/docs/source/index.md index 85bd5d4..29a76ed 100644 --- a/docs/source/index.md +++ b/docs/source/index.md @@ -1,7 +1,7 @@ # Libby Libby is a tiny messaging library built on [Bamboo](https://github.com/CaltechOpticalObservatories/bamboo) -with pluggable transports (ZMQ or RabbitMQ). It gives you: +with pluggable transports (ZMQ, RabbitMQ or Cyclone DDS). It gives you: - **Keywords** — typed, named values (`show` / `modify`) served over RPC, with a registry, auto-generated `keys.list` / `keys.describe` / `keys.read` diff --git a/docs/source/installation.md b/docs/source/installation.md index f4126af..7652f11 100644 --- a/docs/source/installation.md +++ b/docs/source/installation.md @@ -35,6 +35,17 @@ immediately without reinstalling: pip install -e . ``` +For the Cyclone DDS transport, install the `cyclonedds` extra. On Python 3.10 +to 3.13 its wheel bundles the Cyclone C library: + +```bash +pip install -e ".[cyclonedds]" +``` + +cyclonedds 11.0.1 has no Python 3.14 wheel, and its source build fails to +import on 3.14; the fix is on cyclonedds-python's master branch but not yet in +a release. Until one ships, the transport is not installable on 3.14. + To also pull in the docs toolchain (Sphinx + the Shibuya theme): ```bash @@ -54,6 +65,7 @@ python -m unittest discover -s tests ``` Most of `tests/` needs no transport at all. `tests/test_client_integration.py` -is the exception: it starts a real `LibbyDaemon` over RabbitMQ and exercises -`Client` against it, and skips itself automatically if no broker is reachable -at `amqp://localhost`. +is the exception: it starts a real `LibbyDaemon` and exercises `Client` +against it, once per transport. The RabbitMQ cases skip themselves if no +broker is reachable at `amqp://localhost`, and the Cyclone DDS cases unless +the `cyclonedds` extra is installed. diff --git a/libby/cli/cli_config.example.yaml b/libby/cli/cli_config.example.yaml index 0cab995..d1a0d4c 100644 --- a/libby/cli/cli_config.example.yaml +++ b/libby/cli/cli_config.example.yaml @@ -5,11 +5,19 @@ # cp /cli/cli_config.example.yaml ~/.libby/cli_config.yaml # # All keys are optional. Built-in defaults match what's shown here. -# Override per-call with --config / --transport / --rabbitmq-url flags. +# Override per-call with --config / --transport / --rabbitmq-url / +# --dds-domain / --dds-config flags. transport: rabbitmq rabbitmq_url: amqp://localhost +# Cyclone DDS (transport: cyclonedds). Peers are found by discovery; only +# peers on the same domain see each other. dds_config is Cyclone XML, inline +# or a path/URI, e.g. to list unicast peers where multicast is blocked; when +# unset, Cyclone reads CYCLONEDDS_URI. +# dds_domain_id: 0 +# dds_config: file:///etc/libby/cyclonedds.xml + # ZMQ address book (peer_id -> tcp://host:port). Only consulted when # transport=zmq. Leave empty for RabbitMQ deployments. peers: {} diff --git a/libby/cli/libby_cli.py b/libby/cli/libby_cli.py index 4f42d81..33068a2 100644 --- a/libby/cli/libby_cli.py +++ b/libby/cli/libby_cli.py @@ -29,9 +29,13 @@ from libby.config_resolve import ( DEFAULT_BIND, DEFAULT_CONFIG_PATH, + DEFAULT_DDS_DOMAIN, DEFAULT_RABBITMQ_URL, + TRANSPORTS, load_cli_config, resolve_address_book, + resolve_dds_config, + resolve_dds_domain, resolve_rabbitmq_url, resolve_transport, ) @@ -149,6 +153,17 @@ def _mk_libby(namespace: argparse.Namespace, config: Dict[str, Any]) -> Libby: rabbitmq_url=url, keys=[], ) + if transport == "cyclonedds": + domain_id = resolve_dds_domain(namespace.dds_domain, config) + logger.debug( + "connect self_id=%s transport=cyclonedds domain=%d", self_id, domain_id + ) + return Libby.cyclonedds( + self_id=self_id, + keys=[], + domain_id=domain_id, + dds_config=resolve_dds_config(namespace.dds_config, config), + ) address_book = resolve_address_book(config, namespace.addr) bind = namespace.bind or DEFAULT_BIND logger.debug( @@ -597,8 +612,12 @@ def _printer(msg): def _connection_key(namespace: argparse.Namespace, config: Dict[str, Any]) -> str: """Name the broker or address book a completion cache entry belongs to.""" - if resolve_transport(namespace.transport, config) == "rabbitmq": + transport = resolve_transport(namespace.transport, config) + if transport == "rabbitmq": return f"rabbitmq {_redact_url(resolve_rabbitmq_url(namespace.rabbitmq_url, config))}" + if transport == "cyclonedds": + return (f"cyclonedds {resolve_dds_domain(namespace.dds_domain, config)} " + f"{resolve_dds_config(namespace.dds_config, config)}") return f"zmq {sorted(resolve_address_book(config, namespace.addr).items())}" @@ -678,10 +697,15 @@ def build_parser() -> argparse.ArgumentParser: def add_common(p): p.add_argument("--config", help=f"Path to cli_config.yaml (default: {DEFAULT_CONFIG_PATH})") - p.add_argument("--transport", choices=("zmq", "rabbitmq"), + p.add_argument("--transport", choices=TRANSPORTS, help="Override transport (default: from config or 'rabbitmq')") p.add_argument("--rabbitmq-url", help=f"Override RabbitMQ URL (default: from config or '{DEFAULT_RABBITMQ_URL}')") + p.add_argument("--dds-domain", type=int, + help=f"Override Cyclone DDS domain id (default: from config or {DEFAULT_DDS_DOMAIN})") + p.add_argument("--dds-config", + help="Override Cyclone XML config, inline or a path/URI " + "(default: from config, else CYCLONEDDS_URI)") p.add_argument("--bind", help=f"ZMQ ROUTER bind (default: {DEFAULT_BIND})") p.add_argument("--addr", action="append", metavar="peer=tcp://host:port", diff --git a/libby/client.py b/libby/client.py index f5ad646..447cc81 100644 --- a/libby/client.py +++ b/libby/client.py @@ -13,8 +13,11 @@ from .config_resolve import ( DEFAULT_BIND, + DEFAULT_DDS_DOMAIN, load_cli_config, resolve_address_book, + resolve_dds_config, + resolve_dds_domain, resolve_rabbitmq_url, resolve_transport, ) @@ -95,6 +98,18 @@ def rabbitmq( """Connect over RabbitMQ.""" return cls(Libby.rabbitmq(self_id=self_id, rabbitmq_url=rabbitmq_url, keys=[])) + @classmethod + def cyclonedds( + cls, + *, + self_id: str = DEFAULT_SELF_ID, + domain_id: int = DEFAULT_DDS_DOMAIN, + dds_config: Optional[str] = None, + ) -> "Client": + """Connect over Cyclone DDS; peers are found by discovery.""" + return cls(Libby.cyclonedds(self_id=self_id, keys=[], domain_id=domain_id, + dds_config=dds_config)) + @classmethod def zmq( cls, @@ -123,9 +138,14 @@ def from_config( ) -> "Client": """Build a Client from cli_config.yaml, resolving transport as the CLI does.""" config = load_cli_config(path) - if resolve_transport(None, config) == "rabbitmq": + transport = resolve_transport(None, config) + if transport == "rabbitmq": return cls.rabbitmq(self_id=self_id, rabbitmq_url=resolve_rabbitmq_url(None, config)) + if transport == "cyclonedds": + return cls.cyclonedds(self_id=self_id, + domain_id=resolve_dds_domain(None, config), + dds_config=resolve_dds_config(None, config)) return cls.zmq(self_id=self_id, address_book=resolve_address_book(config)) def show(self, name: str, *, timeout_s: float = DEFAULT_TIMEOUT_S) -> Dict[str, Any]: diff --git a/libby/config_resolve.py b/libby/config_resolve.py index 0a0492a..e934d50 100644 --- a/libby/config_resolve.py +++ b/libby/config_resolve.py @@ -14,6 +14,8 @@ DEFAULT_BIND = "tcp://127.0.0.1:56001" DEFAULT_RABBITMQ_URL = "amqp://localhost" DEFAULT_TRANSPORT = "rabbitmq" +DEFAULT_DDS_DOMAIN = 0 +TRANSPORTS = ("zmq", "rabbitmq", "cyclonedds") DEFAULT_CONFIG_PATH = Path.home() / ".libby" / "cli_config.yaml" @@ -33,7 +35,7 @@ def load_cli_config(path: Optional[str] = None) -> Dict[str, Any]: def resolve_transport(override: Optional[str], config: Dict[str, Any]) -> str: """Pick the transport: override → config → default; validated.""" transport = override or config.get("transport") or DEFAULT_TRANSPORT - if transport not in ("zmq", "rabbitmq"): + if transport not in TRANSPORTS: raise ConfigError(f"invalid transport: {transport}") return transport @@ -43,6 +45,21 @@ def resolve_rabbitmq_url(override: Optional[str], config: Dict[str, Any]) -> str return override or config.get("rabbitmq_url") or DEFAULT_RABBITMQ_URL +def resolve_dds_domain(override: Optional[int], config: Dict[str, Any]) -> int: + """Pick the DDS domain id: override → config → default.""" + if override is not None: + return override + domain = config.get("dds_domain_id", DEFAULT_DDS_DOMAIN) + if isinstance(domain, bool) or not isinstance(domain, int): + raise ConfigError(f"invalid dds_domain_id: {domain!r}") + return domain + + +def resolve_dds_config(override: Optional[str], config: Dict[str, Any]) -> Optional[str]: + """Pick the Cyclone XML config: override → config → none (Cyclone reads CYCLONEDDS_URI).""" + return override or config.get("dds_config") or None + + def parse_addr_kv(entry: str) -> Tuple[str, str]: """Parse a 'peer_id=tcp://host:port' address-book entry.""" if "=" not in entry: diff --git a/libby/cyclonedds_transport.py b/libby/cyclonedds_transport.py new file mode 100644 index 0000000..e2838a9 --- /dev/null +++ b/libby/cyclonedds_transport.py @@ -0,0 +1,263 @@ +"""Cyclone DDS transport: bamboo envelopes as typed DDS samples, peer to peer. + +No ``from __future__ import annotations``: cyclonedds resolves IDL field +types from live annotations. +""" +import json +import threading +import time +from dataclasses import dataclass +from typing import Callable, Dict, Optional, Tuple + +from bamboo import Envelope, Message, MsgType +from cyclonedds.core import (DDSException, GuardCondition, InstanceState, Policy, Qos, + ReadCondition, SampleState, ViewState, WaitSet) +from cyclonedds.domain import Domain, DomainParticipant +from cyclonedds.idl import IdlStruct, types +from cyclonedds.pub import DataWriter, Publisher +from cyclonedds.sub import DataReader, Subscriber +from cyclonedds.topic import Topic +from cyclonedds.util import duration + +from ._transport_errors import describe_exception + +DestStr = str +SrcStr = str + +DIRECT_TOPIC = "libby_direct" +BROADCAST_TOPIC = "libby_broadcast" +MTU_BYTES = 512 * 1024 +MAX_BLOCKING_S = 1 +TAKE_BATCH = 64 +RX_WAKE_S = 1 +# A first write to a peer waits this long for its reader, so it is not lost to discovery +MATCH_TIMEOUT_S = 1.0 +# start() returns once no new peer has been matched for SETTLE_S, or after DISCOVERY_MAX_S +DISCOVERY_SETTLE_S = 0.1 +DISCOVERY_MAX_S = 1.0 +POLL_S = 0.01 + + +@dataclass +class DdsEnvelope(IdlStruct, typename="libby.Envelope"): # pylint: disable=too-many-instance-attributes + """bamboo's ``Envelope`` plus the message's binary part, as one DDS sample. + + Absent optional strings travel as ``""`` and an absent binary as empty. + """ + + version: types.int32 + type: str + transid: str + key: str + payload: str + time: str + destid: str + sourceid: str + traceparent: str + tracestate: str + binary: types.sequence[types.uint8] + + +def to_sample(msg: Message) -> DdsEnvelope: + """Encode a bamboo message as a DDS sample.""" + env = msg.env + return DdsEnvelope( + version=env.version, + type=str(env.type), + transid=env.transid, + key=env.key or "", + payload=json.dumps(env.payload, separators=(",", ":")), + time=env.time, + destid=env.destid or "", + sourceid=env.sourceid, + traceparent=env.traceparent or "", + tracestate=env.tracestate or "", + binary=msg.binary or b"", + ) + + +def to_message(sample: DdsEnvelope) -> Message: + """Decode a DDS sample back into a bamboo message.""" + envelope = Envelope( + version=sample.version, + type=MsgType(sample.type), + transid=sample.transid, + key=sample.key or None, + payload=json.loads(sample.payload), + time=sample.time, + destid=sample.destid or None, + sourceid=sample.sourceid, + traceparent=sample.traceparent or None, + tracestate=sample.tracestate or None, + ) + # An octet sequence reads back as a list of ints + return Message(envelope, bytes(sample.binary) if sample.binary else None) + + +def _qos(*partitions: str) -> Qos: + """Reliable and lossless, optionally restricted to partitions.""" + policies = [Policy.Reliability.Reliable(duration(seconds=MAX_BLOCKING_S)), + Policy.History.KeepAll] + if partitions: + policies.append(Policy.Partition(list(partitions))) + return Qos(*policies) + + +def _matched_readers(writer: DataWriter) -> int: + """Count the readers a writer is matched with. + + ``get_matched_subscriptions()`` raises IndexError when a reader matches + between its sizing and filling calls (cyclonedds-python 11.0.1). + """ + return writer.get_publication_matched_status().current_count + + +class CycloneDdsTransport: # pylint: disable=too-many-instance-attributes + """ + Cyclone DDS transport for bamboo, with no broker and no address book. + + Carries ``Message`` objects rather than byte frames (``pack_frames``). + + - Each peer reads ``libby_direct`` in a partition named by its own id; a + direct message is written into the partition named by its ``destid``, + so DDS delivers it only to that peer. + - Every peer reads ``libby_broadcast``; broadcasts reach the sender too, + as on RabbitMQ. + - Peers find each other through DDS discovery. ``dds_config`` is Cyclone + XML (or a path/URI to it), e.g. to list unicast peers where multicast + is blocked; when unset, Cyclone reads ``CYCLONEDDS_URI``. + """ + + pack_frames = False + + def __init__(self, peer_id: str, domain_id: int = 0, dds_config: Optional[str] = None, + group_id: Optional[str] = None): + self._peer_id = peer_id + self._group_id = group_id + self._cb: Optional[Callable[[SrcStr, Message], None]] = None + self._stop = threading.Event() + self._rx_thread: Optional[threading.Thread] = None + self._writers: Dict[str, Tuple[DataWriter, threading.Event]] = {} + self._writers_lock = threading.Lock() + try: + # Kept alive for the participant's lifetime when a config is given + self._domain = Domain(domain_id, dds_config) if dds_config else None + self._participant = DomainParticipant(domain_id) + self._direct_topic = Topic(self._participant, DIRECT_TOPIC, DdsEnvelope) + broadcast_topic = Topic(self._participant, BROADCAST_TOPIC, DdsEnvelope) + self._direct_reader = DataReader( + Subscriber(self._participant, qos=_qos(peer_id)), self._direct_topic, qos=_qos()) + self._broadcast_reader = DataReader( + Subscriber(self._participant), broadcast_topic, qos=_qos()) + self._broadcast_writer = DataWriter( + Publisher(self._participant), broadcast_topic, qos=_qos()) + self._waitset = WaitSet(self._participant) + self._wake = GuardCondition(self._participant) + any_sample = SampleState.Any | ViewState.Any | InstanceState.Any + for reader in (self._direct_reader, self._broadcast_reader): + self._waitset.attach(ReadCondition(reader, any_sample)) + self._waitset.attach(self._wake) + except DDSException as e: + raise RuntimeError( + f"Failed to setup Cyclone DDS transport: {describe_exception(e)}" + ) from e + + @property + def group_id(self) -> Optional[str]: + """Group this peer belongs to, as given at construction.""" + return self._group_id + + @property + def mtu(self) -> int: + """Maximum message size; DDS fragments large samples itself.""" + return MTU_BYTES + + def start(self) -> None: + """Start receiving, then wait (bounded) for discovery to settle. + + Returning only once already-running peers are matched means a first + broadcast, such as a ``keys.list`` sweep, reaches them. + """ + if self._rx_thread and self._rx_thread.is_alive(): + return + self._stop.clear() + self._rx_thread = threading.Thread(target=self._rx_loop, daemon=True, + name="libby-dds-rx") + self._rx_thread.start() + self._settle_discovery() + + def stop(self) -> None: + """Stop receiving and drop later sends.""" + self._stop.set() + self._wake.set(True) + if self._rx_thread: + self._rx_thread.join(timeout=2.0) + + def on_receive(self, cb: Callable[[SrcStr, Message], None]) -> None: + """Register ``cb(source, message)``; ``source`` is ``"peer:"``.""" + self._cb = cb + + def send(self, dest: DestStr, msg: Message) -> None: + """Send to ``"peer:"`` or ``"broadcast:*"``; dropped once stopped. + + Delivery errors are dropped silently, matching bamboo's no-NACK policy. + """ + if self._stop.is_set(): + return + writer = (self._broadcast_writer if dest.startswith("broadcast:") + else self._direct_writer(dest.split(":", 1)[-1])) + try: + writer.write(to_sample(msg)) + except DDSException: + pass + + def _direct_writer(self, peer_id: str) -> DataWriter: + """Return the cached writer for a peer, creating it on first use. + + Only the first use waits for the peer's reader, and it waits outside + the lock so sends to other peers are not held up; concurrent first + sends to the same peer wait on that same match. + """ + with self._writers_lock: + entry = self._writers.get(peer_id) + created = entry is None + if created: + writer = DataWriter(Publisher(self._participant, qos=_qos(peer_id)), + self._direct_topic, qos=_qos()) + entry = self._writers[peer_id] = (writer, threading.Event()) + writer, matched = entry + if created: + deadline = time.monotonic() + MATCH_TIMEOUT_S + while not _matched_readers(writer) and time.monotonic() < deadline: + time.sleep(POLL_S) + matched.set() + else: + matched.wait(MATCH_TIMEOUT_S) + return writer + + def _settle_discovery(self) -> None: + start = time.monotonic() + matched = _matched_readers(self._broadcast_writer) + changed = start + while (now := time.monotonic()) - start < DISCOVERY_MAX_S: + if now - changed >= DISCOVERY_SETTLE_S: + return + time.sleep(POLL_S) + current = _matched_readers(self._broadcast_writer) + if current != matched: + matched, changed = current, time.monotonic() + + def _rx_loop(self) -> None: + while not self._stop.is_set(): + self._waitset.wait(duration(seconds=RX_WAKE_S)) + for reader in (self._direct_reader, self._broadcast_reader): + for sample in reader.take(N=TAKE_BATCH): + if self._stop.is_set(): + return + if not isinstance(sample, DdsEnvelope) or self._cb is None: + continue + try: + message = to_message(sample) + except ValueError: + continue # malformed payload or type, dropped like any undeliverable frame + self._cb(f"peer:{message.env.sourceid}", message) diff --git a/libby/daemon.py b/libby/daemon.py index 4f11a44..d00fc61 100644 --- a/libby/daemon.py +++ b/libby/daemon.py @@ -73,6 +73,8 @@ class LibbyDaemon: "discovery_interval_s", "transport", "rabbitmq_url", + "dds_domain_id", + "dds_config", "group_id", "fail_fast_on_start", } @@ -88,6 +90,8 @@ class LibbyDaemon: # their preferred transport explicitly without changing behavior for existing ZMQ users. transport: str = "zmq" rabbitmq_url: Optional[str] = None + dds_domain_id: int = 0 + dds_config: Optional[str] = None group_id: Optional[str] = None # A daemon that fails to initialize should not continue advertising a @@ -277,6 +281,12 @@ def config_address_book(self) -> Dict[str, str]: def config_rabbitmq_url(self) -> str: return self.rabbitmq_url or "amqp://localhost" + def config_dds_domain_id(self) -> int: + return int(self.dds_domain_id) + + def config_dds_config(self) -> Optional[str]: + return self.dds_config + def config_group_id(self) -> Optional[str]: return self.group_id @@ -363,6 +373,17 @@ def build_libby(self) -> Libby: is_daemon=True, ) + if transport == "cyclonedds": + return Libby.cyclonedds( + self_id=self.config_peer_id(), + keys=[], + callback=None, + group_id=self.config_group_id(), + domain_id=self.config_dds_domain_id(), + dds_config=self.config_dds_config(), + is_daemon=True, + ) + if transport == "zmq": return Libby.zmq( self_id=self.config_peer_id(), @@ -379,7 +400,7 @@ def build_libby(self) -> Libby: raise ValueError( f"unsupported Libby transport {self.transport!r}; " - "expected 'zmq' or 'rabbitmq'" + "expected 'zmq', 'rabbitmq' or 'cyclonedds'" ) def start(self) -> None: diff --git a/libby/how_to_create_a_daemon.md b/libby/how_to_create_a_daemon.md index 9dc05a7..4ad159e 100644 --- a/libby/how_to_create_a_daemon.md +++ b/libby/how_to_create_a_daemon.md @@ -2,7 +2,7 @@ This guide shows you how to create a peer using the **Libby** and the **`LibbyDaemon`** base class. You'll wire a transport, discovery, RPC handlers, and pub/sub with just a few overrides. -> Libby is transport-agnostic. Examples below show both **ZMQ** (peer-to-peer) and **RabbitMQ** (brokered) transports. +> Libby is transport-agnostic. Examples below show **ZMQ** (peer-to-peer), **RabbitMQ** (brokered) and **Cyclone DDS** (peer-to-peer with discovery) transports. --- @@ -33,7 +33,7 @@ sudo systemctl start rabbitmq-server ## 1) What `LibbyDaemon` gives you - **Lifecycle**: easy start/stop, and proper termination handling -- **Transport**: choose ZMQ (peer-to-peer) or RabbitMQ (brokered) with a single line +- **Transport**: choose ZMQ (peer-to-peer), RabbitMQ (brokered) or Cyclone DDS (peer-to-peer with discovery) with a single line - **Discovery**: optional periodic HELLO - **RPC (REQ/RESP)**: register keys and handle requests in one method - **Pub/Sub (PUB)**: register event listeners and subscribe to topics @@ -121,6 +121,26 @@ if __name__ == "__main__": ``` **Note:** Currently, you need to manually specify peer keys using `learn_peer_keys()`. Automatic discovery is not yet fully working with either transport. + +### Cyclone DDS Transport (peer-to-peer, discovered) + +```python +from libby.daemon import LibbyDaemon + +class MyPeer(LibbyDaemon): + peer_id = "peer-X" + group_id = "workers" + + # Cyclone DDS config; needs the cyclonedds extra + transport = "cyclonedds" + dds_domain_id = 0 # only peers on the same domain see each other + dds_config = None # Cyclone XML or a path/URI to it; None reads CYCLONEDDS_URI + + # No bind, address_book or broker: DDS discovery finds the peers + +if __name__ == "__main__": + MyPeer().serve() +``` --- ## 3) A simple peer @@ -272,7 +292,7 @@ if __name__ == "__main__": ## 6) Transport Selection -Libby supports two transports that you can switch between with a single line: +Libby supports three transports that you can switch between with a single line: ### ZMQ Transport (default) - **Peer-to-peer**: Direct socket connections between peers @@ -287,7 +307,14 @@ Libby supports two transports that you can switch between with a single line: - **Best for**: Dynamic scaling, easier deployment, built-in monitoring - **Setup**: Install and start RabbitMQ server -**To switch transports**: Just change `transport = "zmq"` to `transport = "rabbitmq"` in your peer class. Everything else stays the same! +### Cyclone DDS Transport +- **Peer-to-peer with discovery**: messages go straight between peers, which find each other through DDS discovery +- **Requires**: the `cyclonedds` extra; optionally `dds_domain_id` and `dds_config` +- **No address book or broker**: a direct message is routed by its destination's DDS partition +- **Best for**: many peers and high request rates without a broker in the path +- **Setup**: `pip install libby[cyclonedds]`; on networks that block multicast, list unicast peers in the Cyclone XML (`dds_config` or `CYCLONEDDS_URI`) + +**To switch transports**: Just change `transport = "zmq"` to `transport = "rabbitmq"` or `transport = "cyclonedds"` in your peer class. Everything else stays the same! ### Group ID (optional) @@ -296,19 +323,20 @@ Optional `group_id` for grouping peers: ```python class MyPeer(LibbyDaemon): peer_id = "worker-1" - transport = "zmq" # or "rabbitmq" + transport = "zmq" # or "rabbitmq" / "cyclonedds" group_id = "workers" # group this peer belongs to ``` - **RabbitMQ**: The group ID is included in the queue name (e.g., `libby.group.workers.peer.worker-1`) and enables group-aware features. - **ZMQ**: The group ID is stored as metadata and can be used for group-aware routing functions (turn on all devices in a group). +- **Cyclone DDS**: The group ID is part of the peer's wire id (`workers.worker-1`), which names its DDS partition. --- ## 7) Design Notes - **No retries** (protocol choice): sender waits for ACK and optionally RESP; you can handle retries at the app level if needed. -- **Transport-agnostic**: Both ZMQ and RabbitMQ implement Bamboo's `Transport` interface. The same peer code works with either transport. +- **Transport-agnostic**: ZMQ, RabbitMQ and Cyclone DDS implement Bamboo's `Transport` interface (Cyclone DDS with `pack_frames = False`, carrying the envelope as typed DDS fields). The same peer code works with any transport. - **Encapsulation goal**: application peers only implement business logic (`on_req`, `on_event`) and a few config methods. - **Handlers are payload-only.** They receive a Python dict and return anything JSON-serializable. If the return is not a dict, LibbyDaemon auto-wraps it as `{"data": }`. diff --git a/libby/libby.py b/libby/libby.py index cf2a7e3..8e132d8 100644 --- a/libby/libby.py +++ b/libby/libby.py @@ -165,6 +165,57 @@ def rabbitmq( is_daemon=is_daemon, ) + @classmethod + def cyclonedds( + cls, + self_id: str, + keys: Optional[List[str]] = None, + callback: Optional[Callable[[dict, dict], Optional[dict]]] = None, + group_id: Optional[str] = None, + *, + domain_id: int = 0, + dds_config: Optional[str] = None, + is_daemon: bool = False, + ) -> "Libby": + """ + Create a Libby instance using the Cyclone DDS transport: peer to peer, + no broker, peers found by DDS discovery. + + Args: + self_id: Unique identifier for this peer + keys: List of RPC keys this peer will serve + callback: Default callback for RPC requests + group_id: Optional group identifier + domain_id: DDS domain; only peers on the same domain see each other + dds_config: Cyclone XML, or a path/URI to it; when unset, Cyclone + reads ``CYCLONEDDS_URI`` + + Returns: + Configured Libby instance + """ + try: + from .cyclonedds_transport import CycloneDdsTransport + except Exception as e: + raise RuntimeError( + "Cyclone DDS transport not available. Install optional deps:\n" + " pip install libby[cyclonedds]" + ) from e + + wire_id = qualified_peer_id(self_id, group_id) + t = CycloneDdsTransport(peer_id=wire_id, domain_id=domain_id, + dds_config=dds_config, group_id=group_id) + t.start() + return cls( + self_id=wire_id, + transport=t, + keys=keys, + callback=callback, + discover=False, + discover_interval_s=0, + hello_on_start=False, + is_daemon=is_daemon, + ) + # lifecycle def start(self) -> None: if hasattr(self.transport, "start"): diff --git a/pyproject.toml b/pyproject.toml index bd9d07a..c5cd13c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,6 +23,9 @@ dependencies = [ ] [project.optional-dependencies] +cyclonedds = [ + "cyclonedds>=11.0.1", +] influxdb = [ "influxdb-client>=1.40", ] diff --git a/tests/test_client_integration.py b/tests/test_client_integration.py index 33e7662..9ae10d9 100644 --- a/tests/test_client_integration.py +++ b/tests/test_client_integration.py @@ -2,15 +2,19 @@ Every case runs once per transport, so a change that only holds over RabbitMQ fails here. The ZMQ cases need no external service and always run; the RabbitMQ -cases skip unless a broker is reachable at ``amqp://localhost``. +cases skip unless a broker is reachable at ``amqp://localhost``; the Cyclone DDS +cases skip unless cyclonedds is installed, and use a random DDS domain so +concurrent runs cannot hear each other. -There are no sleeps or readiness polls anywhere below, deliberately. Both +There are no sleeps or readiness polls anywhere below, deliberately. All transports block in ``start()`` until they can be talked to, and ``test_first_read_needs_no_warmup`` asserts exactly that, so a regression in the barrier fails loudly instead of being absorbed by a retry. """ from __future__ import annotations +import importlib.util +import random import socket import threading import unittest @@ -28,6 +32,8 @@ LISTING_TIMEOUT_S = 2.0 SERVE_STOP_TIMEOUT_S = 10.0 INITIAL_POSITION = 10.0 +DDS_AVAILABLE = importlib.util.find_spec("cyclonedds") is not None +DDS_DOMAIN_ID = random.randint(1, 230) def _broker_available() -> bool: @@ -110,6 +116,22 @@ class _ZmqLifecycleDaemon(_LifecycleDaemon): transport = "zmq" +class _DdsFixtureDaemon(_FixtureDaemon): + """Fixture daemon served over Cyclone DDS.""" + + peer_id = "pickofftestdds" + transport = "cyclonedds" + dds_domain_id = DDS_DOMAIN_ID + + +class _DdsLifecycleDaemon(_LifecycleDaemon): + """Lifecycle daemon served over Cyclone DDS.""" + + peer_id = "lifecycletestdds" + transport = "cyclonedds" + dds_domain_id = DDS_DOMAIN_ID + + def _started_daemon( daemon_cls: Type[_FixtureDaemon], daemon_peer_id: str, @@ -460,5 +482,59 @@ def build_peer(self) -> Tuple[LibbyDaemon, Client, str]: return daemon, client, peer +@unittest.skipUnless(DDS_AVAILABLE, "cyclonedds not installed") +class CycloneDdsClientTests(_Bases.ClientCases): + """Client cases over Cyclone DDS.""" + + @classmethod + def setUpClass(cls): + cls.daemon = _DdsFixtureDaemon() + cls.daemon.start() + cls.peer = f"{GROUP_ID}.{_DdsFixtureDaemon.peer_id}" + cls.client = Client.cyclonedds(domain_id=DDS_DOMAIN_ID) + + @classmethod + def tearDownClass(cls): + cls.client.close() + cls.daemon.stop() + + +@unittest.skipUnless(DDS_AVAILABLE, "cyclonedds not installed") +class CycloneDdsPeerListingTests(_Bases.PeerListingCases): + """Peer listing cases over Cyclone DDS, where discovery finds every daemon.""" + + daemons: List[LibbyDaemon] + + @classmethod + def setUpClass(cls): + cls.daemons = [ + _started_daemon(_DdsFixtureDaemon, "listingonedds", GROUP_ID), + _started_daemon(_DdsFixtureDaemon, "listingtwodds", GROUP_ID), + _started_daemon(_DdsFixtureDaemon, "listingotherdds", OTHER_GROUP_ID), + ] + cls.group_peers = (f"{GROUP_ID}.listingonedds", f"{GROUP_ID}.listingtwodds") + cls.other_peer = f"{OTHER_GROUP_ID}.listingotherdds" + cls.impostor_peer = f"{GROUP_ID}.impostordds" + cls.impostor = Client.cyclonedds(self_id=cls.impostor_peer, domain_id=DDS_DOMAIN_ID) + cls.client = Client.cyclonedds(domain_id=DDS_DOMAIN_ID) + + @classmethod + def tearDownClass(cls): + cls.client.close() + cls.impostor.close() + for daemon in cls.daemons: + daemon.stop() + + +@unittest.skipUnless(DDS_AVAILABLE, "cyclonedds not installed") +class CycloneDdsLifecycleTests(_Bases.LifecycleCases): + """Lifecycle cases over Cyclone DDS.""" + + def build_peer(self) -> Tuple[LibbyDaemon, Client, str]: + return (_DdsLifecycleDaemon(), + Client.cyclonedds(domain_id=DDS_DOMAIN_ID), + f"{GROUP_ID}.{_DdsLifecycleDaemon.peer_id}") + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_cyclonedds_transport.py b/tests/test_cyclonedds_transport.py new file mode 100644 index 0000000..5093907 --- /dev/null +++ b/tests/test_cyclonedds_transport.py @@ -0,0 +1,124 @@ +"""Unit tests for CycloneDdsTransport: envelope encoding and delivery. + +Delivery cases run real participants on a random DDS domain, so concurrent +runs cannot hear each other. Skipped when cyclonedds is not installed. +""" +from __future__ import annotations + +import dataclasses +import queue +import random +import unittest +from typing import List, Tuple + +from bamboo import Envelope, Message, MessageBuilder + +try: + from libby.cyclonedds_transport import CycloneDdsTransport, DdsEnvelope, to_message, to_sample +except ImportError: # cyclonedds is an optional extra + CycloneDdsTransport = None + +TRACEPARENT = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01" +DELIVERY_TIMEOUT_S = 5.0 +SILENCE_S = 0.5 +MAX_DOMAIN_ID = 230 + + +def _wire_round_trip(msg: Message) -> Message: + """Encode, serialize as DDS would, and decode.""" + return to_message(DdsEnvelope.deserialize(to_sample(msg).serialize())) + + +def _request(source: str, dest: str) -> Message: + return MessageBuilder(source).req("positionvalue", {"value": 1.5}).to(dest).build() + + +@unittest.skipIf(CycloneDdsTransport is None, "cyclonedds not installed") +class EnvelopeEncodingTests(unittest.TestCase): + """bamboo messages survive the trip through the DDS sample type.""" + + def test_every_field_round_trips(self): + """Carry every envelope field, the payload and the binary part unchanged.""" + msg = (MessageBuilder("hsfei.client").req("positionvalue", {"value": [1, "a", None]}) + .to("hsfei.pickoff").trace(TRACEPARENT, "vendor=1") + .binary(b"\x00\n\n\xff").build()) + self.assertEqual(_wire_round_trip(msg), msg) + + def test_absent_optional_fields_stay_absent(self): + """Decode empty strings and an empty binary back to None.""" + msg = MessageBuilder("hsfei.pickoff").pub("alerts", {}).to(None).build() + decoded = _wire_round_trip(msg) + self.assertIsNone(decoded.env.destid) + self.assertIsNone(decoded.env.traceparent) + self.assertIsNone(decoded.binary) + self.assertEqual(decoded, msg) + + def test_sample_type_covers_every_envelope_field(self): + """Fail when bamboo adds an envelope field the DDS type does not carry.""" + envelope_fields = {field.name for field in dataclasses.fields(Envelope)} + sample_fields = {field.name for field in dataclasses.fields(DdsEnvelope)} + self.assertEqual(sample_fields, envelope_fields | {"binary"}) + + +@unittest.skipIf(CycloneDdsTransport is None, "cyclonedds not installed") +class DeliveryTests(unittest.TestCase): + """Direct messages reach only their addressee; broadcasts reach everyone.""" + + def setUp(self) -> None: + domain_id = random.randint(1, MAX_DOMAIN_ID) + self.inboxes: dict[str, "queue.Queue[Tuple[str, Message]]"] = {} + self.transports = {} + for peer_id in ("hsfei.client", "hsfei.pickoff", "hsfei.adc"): + transport = CycloneDdsTransport(peer_id, domain_id=domain_id) + inbox: "queue.Queue[Tuple[str, Message]]" = queue.Queue() + transport.on_receive(lambda source, msg, inbox=inbox: inbox.put((source, msg))) + self.inboxes[peer_id] = inbox + self.transports[peer_id] = transport + for transport in self.transports.values(): + transport.start() + + def tearDown(self) -> None: + for transport in self.transports.values(): + transport.stop() + + def _received(self, peer_id: str, timeout_s: float) -> List[Tuple[str, Message]]: + inbox = self.inboxes[peer_id] + try: + first = inbox.get(timeout=timeout_s) + except queue.Empty: + return [] + return [first] + [inbox.get_nowait() for _ in range(inbox.qsize())] + + def test_direct_message_reaches_its_addressee(self): + """Deliver a direct message with its source and envelope intact.""" + msg = _request("hsfei.client", "hsfei.pickoff") + self.transports["hsfei.client"].send("peer:hsfei.pickoff", msg) + self.assertEqual(self._received("hsfei.pickoff", DELIVERY_TIMEOUT_S), + [("peer:hsfei.client", msg)]) + + def test_direct_message_reaches_no_one_else(self): + """Keep a direct message out of other peers' partitions.""" + msg = _request("hsfei.client", "hsfei.pickoff") + self.transports["hsfei.client"].send("peer:hsfei.pickoff", msg) + self._received("hsfei.pickoff", DELIVERY_TIMEOUT_S) + self.assertEqual(self._received("hsfei.adc", SILENCE_S), []) + self.assertEqual(self._received("hsfei.client", SILENCE_S), []) + + def test_broadcast_reaches_every_peer_including_the_sender(self): + """Deliver a broadcast everywhere, as the RabbitMQ fanout does.""" + msg = MessageBuilder("hsfei.client").pub("alerts", {"level": 1}).to(None).build() + self.transports["hsfei.client"].send("broadcast:*", msg) + for peer_id in self.transports: + self.assertEqual(self._received(peer_id, DELIVERY_TIMEOUT_S), + [("peer:hsfei.client", msg)], peer_id) + + def test_send_after_stop_is_dropped(self): + """Drop a late send rather than raising or delivering it.""" + self.transports["hsfei.client"].stop() + self.transports["hsfei.client"].send( + "peer:hsfei.pickoff", _request("hsfei.client", "hsfei.pickoff")) + self.assertEqual(self._received("hsfei.pickoff", SILENCE_S), []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tox.ini b/tox.ini index 7784b73..9b6c7fd 100644 --- a/tox.ini +++ b/tox.ini @@ -5,15 +5,18 @@ isolated_build = true [testenv] description = Run the test suite -# Installed so the InfluxDB sink and telemetry tests run rather than skip themselves -extras = influxdb, otel +# Installed so the InfluxDB sink, telemetry and Cyclone DDS tests run rather than skip themselves +extras = influxdb, otel, cyclonedds deps = commands = python -m unittest discover -s tests [testenv:docs] description = Build the Sphinx documentation (warnings fail the build) -extras = docs +# cyclonedds so the API reference can import the transport module +extras = + docs + cyclonedds commands = sphinx-build -b html -W --keep-going docs/source docs/build/html From d45597640c5cdd03446d4b500a50c36fe65c41a8 Mon Sep 17 00:00:00 2001 From: Mike Langmayr Date: Thu, 5 Nov 2026 17:57:49 -0800 Subject: [PATCH 2/2] Delete the DDS participant on stop, refuse restart, and survive callback errors --- libby/cyclonedds_transport.py | 45 ++++++++++++++++++++++-------- tests/test_cyclonedds_transport.py | 38 ++++++++++++++++++++++++- 2 files changed, 71 insertions(+), 12 deletions(-) diff --git a/libby/cyclonedds_transport.py b/libby/cyclonedds_transport.py index e2838a9..43b5a49 100644 --- a/libby/cyclonedds_transport.py +++ b/libby/cyclonedds_transport.py @@ -4,6 +4,7 @@ types from live annotations. """ import json +import logging import threading import time from dataclasses import dataclass @@ -21,6 +22,8 @@ from ._transport_errors import describe_exception +logger = logging.getLogger(__name__) + DestStr = str SrcStr = str @@ -136,6 +139,7 @@ def __init__(self, peer_id: str, domain_id: int = 0, dds_config: Optional[str] = self._group_id = group_id self._cb: Optional[Callable[[SrcStr, Message], None]] = None self._stop = threading.Event() + self._closed = False self._rx_thread: Optional[threading.Thread] = None self._writers: Dict[str, Tuple[DataWriter, threading.Event]] = {} self._writers_lock = threading.Lock() @@ -176,22 +180,34 @@ def start(self) -> None: """Start receiving, then wait (bounded) for discovery to settle. Returning only once already-running peers are matched means a first - broadcast, such as a ``keys.list`` sweep, reaches them. + broadcast, such as a ``keys.list`` sweep, reaches them. A stopped + transport cannot be restarted; build a new one. """ + if self._closed: + raise RuntimeError("CycloneDdsTransport was stopped; create a new transport") if self._rx_thread and self._rx_thread.is_alive(): return - self._stop.clear() self._rx_thread = threading.Thread(target=self._rx_loop, daemon=True, name="libby-dds-rx") self._rx_thread.start() self._settle_discovery() def stop(self) -> None: - """Stop receiving and drop later sends.""" + """Stop receiving, drop later sends, and leave the DDS domain. + + Deleting the participant deletes its readers and writers too, so + other peers stop matching this one. + """ + if self._closed: + return + self._closed = True self._stop.set() self._wake.set(True) if self._rx_thread: self._rx_thread.join(timeout=2.0) + # cyclonedds-python has no public close, and `del` would not delete while + # child entities hold references; __del__ deletes the entity now + self._participant.__del__() # pylint: disable=unnecessary-dunder-call def on_receive(self, cb: Callable[[SrcStr, Message], None]) -> None: """Register ``cb(source, message)``; ``source`` is ``"peer:"``.""" @@ -204,12 +220,12 @@ def send(self, dest: DestStr, msg: Message) -> None: """ if self._stop.is_set(): return - writer = (self._broadcast_writer if dest.startswith("broadcast:") - else self._direct_writer(dest.split(":", 1)[-1])) try: + writer = (self._broadcast_writer if dest.startswith("broadcast:") + else self._direct_writer(dest.split(":", 1)[-1])) writer.write(to_sample(msg)) except DDSException: - pass + pass # includes a send racing stop() as the participant is deleted def _direct_writer(self, peer_id: str) -> DataWriter: """Return the cached writer for a peer, creating it on first use. @@ -227,10 +243,12 @@ def _direct_writer(self, peer_id: str) -> DataWriter: entry = self._writers[peer_id] = (writer, threading.Event()) writer, matched = entry if created: - deadline = time.monotonic() + MATCH_TIMEOUT_S - while not _matched_readers(writer) and time.monotonic() < deadline: - time.sleep(POLL_S) - matched.set() + try: + deadline = time.monotonic() + MATCH_TIMEOUT_S + while not _matched_readers(writer) and time.monotonic() < deadline: + time.sleep(POLL_S) + finally: + matched.set() else: matched.wait(MATCH_TIMEOUT_S) return writer @@ -260,4 +278,9 @@ def _rx_loop(self) -> None: message = to_message(sample) except ValueError: continue # malformed payload or type, dropped like any undeliverable frame - self._cb(f"peer:{message.env.sourceid}", message) + try: + self._cb(f"peer:{message.env.sourceid}", message) + except Exception: # pylint: disable=broad-exception-caught + # One bad message must not stop this peer receiving + logger.exception("receive callback failed for a message from %s", + message.env.sourceid) diff --git a/tests/test_cyclonedds_transport.py b/tests/test_cyclonedds_transport.py index 5093907..49ecab8 100644 --- a/tests/test_cyclonedds_transport.py +++ b/tests/test_cyclonedds_transport.py @@ -8,13 +8,15 @@ import dataclasses import queue import random +import time import unittest from typing import List, Tuple from bamboo import Envelope, Message, MessageBuilder try: - from libby.cyclonedds_transport import CycloneDdsTransport, DdsEnvelope, to_message, to_sample + from libby.cyclonedds_transport import (CycloneDdsTransport, DdsEnvelope, _matched_readers, + to_message, to_sample) except ImportError: # cyclonedds is an optional extra CycloneDdsTransport = None @@ -112,6 +114,40 @@ def test_broadcast_reaches_every_peer_including_the_sender(self): self.assertEqual(self._received(peer_id, DELIVERY_TIMEOUT_S), [("peer:hsfei.client", msg)], peer_id) + def test_stop_leaves_the_domain(self): + """Stop being matched by other peers once stopped.""" + # Matching is only observable on the transport's own writer + writer = self.transports["hsfei.pickoff"]._broadcast_writer # pylint: disable=protected-access + before = _matched_readers(writer) + self.transports["hsfei.client"].stop() + deadline = time.monotonic() + DELIVERY_TIMEOUT_S + while _matched_readers(writer) == before and time.monotonic() < deadline: + time.sleep(0.01) + self.assertEqual(_matched_readers(writer), before - 1) + + def test_start_after_stop_is_refused(self): + """Refuse to restart a transport whose participant is gone.""" + self.transports["hsfei.client"].stop() + with self.assertRaises(RuntimeError): + self.transports["hsfei.client"].start() + + def test_failing_callback_does_not_stop_receiving(self): + """Keep receiving after the callback raises for one message.""" + received: List[Message] = [] + + def fail_once(_source: str, msg: Message) -> None: + if not received: + received.append(msg) + raise ValueError("handler bug") + self.inboxes["hsfei.adc"].put(("", msg)) + + self.transports["hsfei.adc"].on_receive(fail_once) + with self.assertLogs("libby.cyclonedds_transport", "ERROR"): + for _ in range(2): + self.transports["hsfei.client"].send( + "peer:hsfei.adc", _request("hsfei.client", "hsfei.adc")) + self.assertEqual(len(self._received("hsfei.adc", DELIVERY_TIMEOUT_S)), 1) + def test_send_after_stop_is_dropped(self): """Drop a late send rather than raising or delivering it.""" self.transports["hsfei.client"].stop()