Create DDS Transport Protocol

dimos/core/__init__.py

@@ -14,6 +14,7 @@
 from dimos.core.rpc_client import RPCClient
 from dimos.core.stream import In, Out, RemoteIn, RemoteOut, Transport
 from dimos.core.transport import (
+    DDSTransport,
     LCMTransport,
     SHMTransport,
     ZenohTransport,
@@ -36,6 +37,7 @@
     "LCMRPC",
     "LCMTF",
     "TF",
+    "DDSTransport",
     "DimosCluster",
     "In",
     "LCMTransport",

dimos/core/transport.py

@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 
+import threading
 from typing import Any, TypeVar
 
 import dimos.core.colors as colors
@@ -27,6 +28,7 @@
 
 from dimos.core.stream import In, Out, Stream, Transport
 from dimos.msgs.protocol import DimosMsg
+from dimos.protocol.pubsub.ddspubsub import DDS, Topic as DDSTopic
 from dimos.protocol.pubsub.impl.jpeg_shm import JpegSharedMemory
 from dimos.protocol.pubsub.impl.lcmpubsub import LCM, JpegLCM, PickleLCM, Topic as LCMTopic
 from dimos.protocol.pubsub.impl.rospubsub import DimosROS, ROSTopic
@@ -258,4 +260,36 @@ def stop(self) -> None:
             self._ros = None
 
 
+class DDSTransport(PubSubTransport[T]):
+    def __init__(self, topic: str, type: type, **kwargs) -> None:  # type: ignore[no-untyped-def]
+        super().__init__(DDSTopic(topic, type))
+        self.dds = DDS(**kwargs)
+        self._started: bool = False
+        self._start_lock = threading.RLock()
+
+    def start(self) -> None:
+        with self._start_lock:
+            if not self._started:
+                self.dds.start()
+                self._started = True
+
+    def stop(self) -> None:
+        with self._start_lock:
+            if self._started:
+                self.dds.stop()
+                self._started = False
+
+    def broadcast(self, _, msg) -> None:
+        with self._start_lock:
+            if not self._started:
+                self.start()
+            self.dds.publish(self.topic, msg)
+
+    def subscribe(self, callback: Callable[[T], None], selfstream: In[T] = None) -> None:  # type: ignore[assignment, override]
+        with self._start_lock:
+            if not self._started:
+                self.start()
+            return self.dds.subscribe(self.topic, lambda msg, topic: callback(msg))  # type: ignore[return-value]
+
+
 class ZenohTransport(PubSubTransport[T]): ...

dimos/protocol/pubsub/__init__.py

@@ -1,9 +1,11 @@
+import dimos.protocol.pubsub.ddspubsub as dds
 import dimos.protocol.pubsub.impl.lcmpubsub as lcm
 from dimos.protocol.pubsub.impl.memory import Memory
 from dimos.protocol.pubsub.spec import PubSub
 
 __all__ = [
     "Memory",
     "PubSub",
+    "dds",
     "lcm",
 ]

dimos/protocol/pubsub/benchmark/testdata.py

@@ -14,12 +14,21 @@
 
 from collections.abc import Generator
 from contextlib import contextmanager
+from dataclasses import dataclass
 from typing import Any
 
+from cyclonedds.idl import IdlStruct
+from cyclonedds.idl.types import sequence, uint8
 import numpy as np
 
 from dimos.msgs.sensor_msgs.Image import Image, ImageFormat
 from dimos.protocol.pubsub.benchmark.type import Case
+from dimos.protocol.pubsub.ddspubsub import (
+    DDS,
+    HIGH_THROUGHPUT_QOS,
+    RELIABLE_QOS,
+    Topic as DDSTopic,
+)
 from dimos.protocol.pubsub.impl.lcmpubsub import LCM, LCMPubSubBase, Topic as LCMTopic
 from dimos.protocol.pubsub.impl.memory import Memory
 from dimos.protocol.pubsub.impl.shmpubsub import (
@@ -174,6 +183,52 @@ def shm_lcm_pubsub_channel() -> Generator[LCMSharedMemory, None, None]:
 )
 
 
+@dataclass
+class DDSBenchmarkData(IdlStruct):
+    """DDS message type for benchmarking with variable-size byte payload."""
+
+    data: sequence[uint8]  # type: ignore[type-arg]
+
+
+@contextmanager
+def dds_high_throughput_pubsub_channel() -> Generator[DDS, None, None]:
+    """DDS with HIGH_THROUGHPUT_QOS - BestEffort, optimized for speed."""
+    dds_pubsub = DDS(qos=HIGH_THROUGHPUT_QOS)
+    dds_pubsub.start()
+    yield dds_pubsub
+    dds_pubsub.stop()
+
+
+@contextmanager
+def dds_reliable_pubsub_channel() -> Generator[DDS, None, None]:
+    """DDS with RELIABLE_QOS - guaranteed delivery."""
+    dds_pubsub = DDS(qos=RELIABLE_QOS)
+    dds_pubsub.start()
+    yield dds_pubsub
+    dds_pubsub.stop()
+
+
+def dds_msggen(size: int) -> tuple[DDSTopic, DDSBenchmarkData]:
+    """Generate DDS message for benchmark."""
+    topic = DDSTopic(name="benchmark/dds", data_type=DDSBenchmarkData)
+    return (topic, DDSBenchmarkData(data=list(make_data_bytes(size))))
+
+
+testcases.append(
+    Case(
+        pubsub_context=dds_high_throughput_pubsub_channel,
+        msg_gen=dds_msggen,
+    )
+)
+
+testcases.append(
+    Case(
+        pubsub_context=dds_reliable_pubsub_channel,
+        msg_gen=dds_msggen,
+    )
+)
+
+
 try:
     from dimos.protocol.pubsub.impl.redispubsub import Redis
 

dimos/protocol/pubsub/ddspubsub.py

@@ -0,0 +1,178 @@
+# Copyright 2025-2026 Dimensional Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+import threading
+from typing import TYPE_CHECKING, Any, TypeAlias
+
+from cyclonedds.core import Listener
+from cyclonedds.pub import DataWriter as DDSDataWriter
+from cyclonedds.qos import Policy, Qos
+from cyclonedds.sub import DataReader as DDSDataReader
+from cyclonedds.topic import Topic as DDSTopic
+
+from dimos.protocol.pubsub.spec import PubSub
+from dimos.protocol.service.ddsservice import DDSService
+from dimos.utils.logging_config import setup_logger
+
+if TYPE_CHECKING:
+    from cyclonedds.idl import IdlStruct
+
+logger = setup_logger()
+
+
+# High-throughput QoS preset
+HIGH_THROUGHPUT_QOS = Qos(
+    Policy.Reliability.BestEffort,
+    Policy.History.KeepLast(depth=1),
+    Policy.Durability.Volatile,
+)
+
+# Reliable QoS preset
+RELIABLE_QOS = Qos(
+    Policy.Reliability.Reliable(max_blocking_time=0),
+    Policy.History.KeepLast(depth=5000),
+    Policy.Durability.Volatile,
+)
+
+
+@dataclass(frozen=True)
+class Topic:
+    """Represents a DDS topic."""
+
+    name: str
+    data_type: type[IdlStruct]
+
+    def __str__(self) -> str:
+        return f"{self.name}#{self.data_type.__name__}"
+
+
+MessageCallback: TypeAlias = Callable[[Any, Topic], None]
+
+
+class _DDSMessageListener(Listener):
+    """Listener for DataReader that dispatches messages to callbacks."""
+
+    __slots__ = ("_callbacks", "_lock", "_topic")
+
+    def __init__(self, topic: Topic) -> None:
+        super().__init__()
+        self._topic = topic
+        self._callbacks: tuple[MessageCallback, ...] = ()
+        self._lock = threading.Lock()
+
+    def add_callback(self, callback: MessageCallback) -> None:
+        """Add a callback to the listener."""
+        with self._lock:
+            self._callbacks = (*self._callbacks, callback)
+
+    def remove_callback(self, callback: MessageCallback) -> None:
+        """Remove a callback from the listener."""
+        with self._lock:
+            self._callbacks = tuple(cb for cb in self._callbacks if cb is not callback)
+
+    def on_data_available(self, reader: DDSDataReader[Any]) -> None:
+        """Called when data is available on the reader."""
+        try:
+            samples = reader.take()
+        except Exception as e:
+            logger.error(f"Error reading from topic {self._topic}: {e}", exc_info=True)
+            return
+        for sample in samples:
+            if sample is not None:
+                for callback in self._callbacks:
+                    try:
+                        callback(sample, self._topic)
+                    except Exception as e:
+                        logger.error(f"Callback error on topic {self._topic}: {e}", exc_info=True)
+
+
+class DDS(DDSService, PubSub[Topic, Any]):
+    def __init__(self, qos: Qos | None = None, **kwargs: Any) -> None:
+        super().__init__(**kwargs)
+        self._qos = qos
+        self._writers: dict[Topic, DDSDataWriter[Any]] = {}
+        self._writer_lock = threading.Lock()
+        self._readers: dict[Topic, DDSDataReader[Any]] = {}
+        self._reader_lock = threading.Lock()
+        self._listeners: dict[Topic, _DDSMessageListener] = {}
+
+    @property
+    def qos(self) -> Qos | None:
+        """Get the QoS settings."""
+        return self._qos
+
+    def _get_writer(self, topic: Topic) -> DDSDataWriter[Any]:
+        """Get or create a DataWriter for the given topic."""
+        with self._writer_lock:
+            if topic not in self._writers:
+                dds_topic = DDSTopic(self.participant, topic.name, topic.data_type)
+                self._writers[topic] = DDSDataWriter(self.participant, dds_topic, qos=self._qos)
+            return self._writers[topic]
+
+    def publish(self, topic: Topic, message: Any) -> None:
+        """Publish a message to a DDS topic."""
+        writer = self._get_writer(topic)
+        try:
+            writer.write(message)
+        except Exception as e:
+            logger.error(f"Error publishing to topic {topic}: {e}", exc_info=True)
+
+    def _get_listener(self, topic: Topic) -> _DDSMessageListener:
+        """Get or create a listener and reader for the given topic."""
+        with self._reader_lock:
+            if topic not in self._readers:
+                dds_topic = DDSTopic(self.participant, topic.name, topic.data_type)
+                listener = _DDSMessageListener(topic)
+                self._readers[topic] = DDSDataReader(
+                    self.participant, dds_topic, qos=self._qos, listener=listener
+                )
+                self._listeners[topic] = listener
+            return self._listeners[topic]
+
+    def subscribe(self, topic: Topic, callback: MessageCallback) -> Callable[[], None]:
+        """Subscribe to a DDS topic with a callback."""
+        listener = self._get_listener(topic)
+        listener.add_callback(callback)
+        return lambda: self._unsubscribe_callback(topic, callback)
+
+    def _unsubscribe_callback(self, topic: Topic, callback: MessageCallback) -> None:
+        """Unsubscribe a callback from a topic."""
+        with self._reader_lock:
+            listener = self._listeners.get(topic)
+        if listener:
+            listener.remove_callback(callback)
+
+    def stop(self) -> None:
+        """Stop the DDS service and clean up resources."""
+        with self._reader_lock:
+            self._readers.clear()
+            self._listeners.clear()
+        with self._writer_lock:
+            self._writers.clear()
+        super().stop()
+
+
+__all__ = [
+    "DDS",
+    "HIGH_THROUGHPUT_QOS",
+    "RELIABLE_QOS",
+    "MessageCallback",
+    "Policy",
+    "Qos",
+    "Topic",
+]