Create DDSTransport DDSPubSubBase

[Kaweees]

Related: #1033, #1058

uv run pytest -svm tool dimos/protocol/pubsub/benchmark/test_benchmark.py --override-ini="addopts=" -k "DDS"
uv run pytest -svm tool dimos/protocol/pubsub/benchmark/test_benchmark.py

[Kaweees]

@greptile

[greptile-apps]

Greptile Overview

Greptile Summary

This PR introduces a DDS (Data Distribution Service) transport layer using CycloneDDS, creating DDSPubSubBase, DDSService, and DDSTransport classes with benchmark tests.

Major changes:

• Implements proper DDS pub/sub with DataWriter/DataReader and listener-based message dispatch via _DDSMessageListener
• Uses lazy initialization for DomainParticipant with double-checked locking for thread safety
• Adds benchmark tests comparing DDS performance with other transports (LCM, shared memory)
• Uses IdlStruct with sequence[uint8] for binary payloads to minimize serialization overhead

Critical issues:

• Attribute mismatch in transport.py:233: DDSTransport.__reduce__ accesses self.topic.dds_type but the Topic class only has typename attribute, causing AttributeError at runtime during serialization
• Duplicate imports and variable definitions in transport.py (lines 17, 23-26, 37)

The implementation correctly addresses previously reported architectural issues by actually publishing to the DDS network and receiving messages through listeners, rather than just using local callbacks.

Confidence Score: 2/5

• This PR has a critical runtime error that will break serialization
• The attribute mismatch in DDSTransport.__reduce__ will cause AttributeError when the transport is serialized (e.g., during pickling for multiprocessing). This is a blocking issue that needs to be fixed before merge.
• Pay close attention to dimos/core/transport.py - the __reduce__ method has incorrect attribute access that will fail at runtime

Important Files Changed

Filename	Overview
dimos/core/transport.py	Added DDSTransport class with incorrect attribute access - uses self.topic.dds_type but Topic class only has typename attribute
dimos/protocol/pubsub/ddspubsub.py	Implements DDSPubSubBase with proper DataWriter/DataReader creation and listener-based message dispatch, resolving previous architectural issues
dimos/protocol/service/ddsservice.py	New DDSService base class with lazy DomainParticipant initialization using double-checked locking pattern

Sequence Diagram

sequenceDiagram
    participant App as Application
    participant DT as DDSTransport
    participant DDS as DDS (DDSPubSubBase)
    participant DS as DDSService
    participant DP as DomainParticipant
    participant DW as DataWriter
    participant DR as DataReader
    participant L as _DDSMessageListener

    Note over App,L: Initialization
    App->>DT: __init__(topic, type)
    DT->>DT: Create Topic(name, typename)
    DT->>DDS: Create DDS instance
    
    App->>DT: start()
    DT->>DDS: start()
    DDS->>DS: get_participant()
    DS->>DP: Create DomainParticipant (lazy init)
    
    Note over App,L: Publishing Flow
    App->>DT: broadcast(msg)
    DT->>DDS: publish(topic, msg)
    DDS->>DDS: _get_writer(topic)
    DDS->>DW: Create DataWriter (if needed)
    DDS->>DW: write(message)
    DW->>DP: Send to DDS network
    
    Note over App,L: Subscription Flow
    App->>DT: subscribe(callback)
    DT->>DDS: subscribe(topic, wrapped_callback)
    DDS->>DDS: _get_reader(topic)
    DDS->>L: Create _DDSMessageListener
    DDS->>DR: Create DataReader with listener
    
    Note over App,L: Message Reception
    DP->>DR: Incoming DDS message
    DR->>L: on_data_available(reader)
    L->>DR: take() samples
    L->>L: Iterate over samples
    L->>App: callback(sample, topic)

Loading

[greptile-apps]

_{2 files reviewed, 2 comments}

_{Edit Code Review Agent Settings | Greptile}

[Kaweees]

@greptile

[greptile-apps]

_{2 files reviewed, 4 comments}

_{Edit Code Review Agent Settings | Greptile}

[Kaweees]

@greptile

[greptile-apps]

_{4 files reviewed, 4 comments}

_{Edit Code Review Agent Settings | Greptile}

[Kaweees]

@greptile

[greptile-apps]

_{5 files reviewed, 3 comments}

_{Edit Code Review Agent Settings | Greptile}

[Kaweees]

@greptile

[greptile-apps]

_{8 files reviewed, 4 comments}

_{Edit Code Review Agent Settings | Greptile}

[Kaweees]

@greptile

[greptile-apps]

_{18 files reviewed, 4 comments}

_{Edit Code Review Agent Settings | Greptile}

[greptile-apps]

logic: Test Message class uses string encoding which breaks with binary data. The make_data() function generates binary bytes, but encoding as latin-1 string will fail for many byte sequences.

DDS IdlStruct should use bytes type directly:

Suggested change

	class Message(IdlStruct):
	"""DDS message with binary data payload following IdlStruct format."""
	payload: str
	def dds_encode(self) -> bytes:
	"""Encode message to bytes for DDS transmission."""
	return self.payload.encode("latin-1")
	@classmethod
	def dds_decode(cls, data: bytes) -> "Message":
	"""Decode bytes back to Message instance."""
	return cls(payload=data.decode("latin-1"))
	@dataclass
	class Message(IdlStruct):
	"""DDS message with binary data payload following IdlStruct format."""
	payload: bytes

Remove the encode/decode methods as CycloneDDS handles serialization of IdlStruct fields automatically.

[Kaweees]

@greptile

[greptile-apps]

_{3 files reviewed, 3 comments}

_{Edit Code Review Agent Settings | Greptile}

[greptile-apps]

Additional Comments (2)

dimos/core/transport.py
TypeVar imported twice

---

dimos/core/transport.py
TypeVar defined twice with conflicting type ignore comments

[paul-nechifor]

This file in on dev now. You might want to rebase onto dev so that it it disappears from the diff of this PR.