Add a Cyclone DDS transport

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 <group>.<daemon>` (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.

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:

docs/source/cli.md

@@ -15,7 +15,7 @@ libby completion bash|zsh                         # print shell code for TAB com
 `<group>.<daemon>` 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).
 

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

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`

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.

libby/cli/cli_config.example.yaml

@@ -5,11 +5,19 @@
 #   cp <libby_pkg>/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: {}

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",

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]:

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: