Retina · Rust · evaluation release

Retina DDS for Rust

DDS is more than publish and subscribe. It combines a data-centric programming model with state for multiple objects, policy-driven matching, bounded histories, discovery, repair, filtering, lifecycle, and status events.

DDS programming models

DDS core

DCPS defines how data is addressed, published, retained, and observed.

Data-centric publish/subscribe

Writers publish application data to topics; readers receive it without naming or connecting to a specific peer.

Participant → Topic<T> → DataWriter<T> / DataReader<T>

Multiple objects on one topic

A field such as a robot ID identifies each object, so DDS can maintain history, ownership, durability, and lifecycle separately for every robot.

write_instance · take_instance · dispose · unregister

State access and event-driven processing

Applications can inspect retained state with read/take operations or react to data and communication-status changes through listeners.

read · take · listener · communication status

Grouped and coherent changes

Publishers and subscribers establish a policy boundary; coherent sets keep readers from observing only part of a related update.

Publisher · Subscriber · begin/end coherent changes

Models built on DDS

Request/reply

A request and its reply are ordinary DDS samples connected by related-sample identity and conventionally paired topics.

request writer + reply reader + correlation identity

ROS 2 execution

Publishers, subscriptions, clients, services, events, and wait sets are mapped onto DDS through the ROS middleware boundary.

retina-rmw

These models do not imply a standalone DDS-RPC client/service facade or a standalone DCPS WaitSet API. The catalog below states Retina's actual public surface.

Retina · Rust

The Retina implementation

Retina exposes Rust DCPS entities with explicit QoS builders and capability crates for discovery, reliability, serialization, shared memory, persistence, ROS integration, and security.

use retina_core::{dds_type, Participant};

dds_type! {
    #[derive(Clone, Copy, Debug, PartialEq)]
    struct Temperature {
        degrees_celsius: f32,
    }
}

let domain_id: u32 = std::env::var("ROS_DOMAIN_ID")
    .ok()
    .and_then(|value| value.parse().ok())
    .unwrap_or(0);
let participant = Participant::new(domain_id);
let topic = participant.topic::<Temperature>("sensors/temperature");
let writer = participant.create_writer(&topic)?;
let reader = participant.create_reader(&topic)?;
writer.write(&Temperature { degrees_celsius: 23.5 })?;
let sample = reader.take()?.expect("temperature sample");

Retina uses Result values. Endpoints unregister automatically on drop. close remains available when the caller needs to observe cleanup errors.

Retina · data and state

Rust feature APIs and algorithms

DCPS entities

Participant::new(domain) · participant.topic::<T>(name) · participant.create_writer(&topic) · participant.create_reader(&topic)

RAII endpoint leasing. Participant-owned leases preserve identity and remove registrations on close or drop.

Local data flow and grouped entity creation with Rust ownership governing lifetime.

let topic = participant.topic::<Temperature>("temperature");
let writer = participant.create_writer(&topic)?;
let reader = participant.create_reader(&topic)?;

Types and bounded serialization

dds_type! { ... } · value.serialize() · writer.write_fixed(&value, &mut storage)

CDR serialization with deterministic type identity. Fixed-size types can serialize into caller-owned storage.

The ordinary writer handles serialization, while fixed paths make allocation explicit and reusable.

dds_type! { struct Temperature { value: f32 } }

let mut storage = [0_u8; Temperature::MAX_SERIALIZED_SIZE];
writer.write_fixed(&sample, &mut storage)?;

Topic instances

writer.write_instance(&value, key), reader.take_instance(key)

Bounded keyed-history selection. The reader selects and compacts the first matching key; retention and ownership state remain per key.

A topic can represent many independently retained logical objects.

writer.write_instance(&state, b"robot-17")?;
let sample = reader.take_instance(b"robot-17")?;

QoS compatibility and partitions

QosProfile::builder().reliable().partition(name).ownership(kind, strength).build()

DDS requested/offered compatibility. Policy ordering and bounded partition wildcard matching admit only compatible endpoints.

Policy choices become matching contracts rather than application-side peer checks.

let qos = QosProfile::builder()
    .reliable()
    .partition("control/*")
    .build();

History and resource limits

QosProfile::builder().keep_last(depth).max_samples(limit).resource_limits(limits).build()

Bounded KEEP_LAST/KEEP_ALL admission. Admission evicts under KEEP_LAST or rejects overflow under KEEP_ALL, including per-instance limits.

Readers and writers carry explicit storage bounds.

let qos = QosProfile::builder()
    .keep_last(8)
    .max_samples(64)
    .build();

Content filters

reader.filter_by(|value| ...) · reader.set_content_filter(filter) · reader.clear_content_filter() · ContentFilter::on_payload(predicate)

Pre-admission content filtering. Deserialized or payload predicates run before reader-history admission and reject malformed data.

Readers receive only the subset they requested and can replace the predicate at runtime.

reader.filter_by(|value: &Temperature| {
    value.degrees_celsius > 80.0
})?;
reader.clear_content_filter()?;

Durability

QosProfile::builder().durability(DurabilityKind::TransientLocal).build() · DurabilityService::open(path) · participant.attach_durability_service(service)

Bounded durability replay. Per-writer and per-key retention preserves original sequence metadata; service policy governs checkpoints.

Late joiners recover retained state under the configured history policy.

let qos = QosProfile::builder()
    .durability(DurabilityKind::TransientLocal)
    .keep_last(8)
    .build();

Time, liveliness, ownership, and order

QosProfile::builder().deadline(duration).lifespan(duration).liveliness(kind, lease).ownership(kind, strength).destination_order(order).build()

Lease expiry, strongest-writer arbitration, and source-order selection. Policy runs at admission or take within configured bounds.

Freshness, authority, activity, and observation order are expressed as DDS policy.

let qos = QosProfile::builder()
    .deadline(Duration::from_millis(10))
    .ownership(OwnershipKind::Exclusive, 100)
    .destination_order(DestinationOrderKind::BySourceTimestamp)
    .build();

Local scope

Coherent changes

publisher.begin_coherent_changes() · publisher.end_coherent_changes() · publisher.discard_coherent_changes()

Publisher-scoped staged commit. Each publisher owns an independent set; readers requesting coherent access receive it only at commit.

Local and shared-memory delivery preserves the set. Network coherent-set identity is not claimed here.

publisher.begin_coherent_changes()?;
trajectory_writer.write(&trajectory)?;
limits_writer.write(&limits)?;
publisher.end_coherent_changes()?;

let qos = QosProfile::builder()
    .reliable()
    .keep_last(32)
    .max_samples(128)
    .deadline(Duration::from_millis(10))
    .partition("control/*")
    .build();

Retina · network and runtime

Rust delivery mechanisms

retina-runtime

Discovery and active runtime

RuntimeParticipant::<T>::build(...) · participant.discover(...) · matched.metrics()

SPDP lease discovery with SEDP requested/offered endpoint matching. Participant leases and endpoint records build a bounded matched path.

The higher-level participant handles ordinary use; runtime types expose active discovery when a harness needs direct control.

let local = RuntimeParticipant::<Temperature>::build(
    identity, "temperature/out", "temperature/in", qos)?;
let matched = local.discover(
    &discovery, &transport, timeout)?;

retina-reliable

Reliability and fragmentation

TypedReliableSession::new(...) · session.write(&value) · session.take_until(deadline) · session.metrics()

RTPS stateful reliable writer/reader repair. DATA/DATA_FRAG, HEARTBEAT/ACKNACK, NACK_FRAG, GAP, and bounded reassembly implement recovery.

Reliable data exchange and explicit protocol metrics under loss and reordering.

let mut session = TypedReliableSession::new(
    &transport, &mut matched, config);
session.write(&sample)?;
let received = session.take_until(deadline)?;

retina-rtps

Wire-level RTPS

parse_rtps_message(bytes) · submessages.next() · write_rtps_message_to(header, submessage, out)

Zero-copy bounded submessage parsing. Borrowed views validate RTPS submessages without taking payload ownership.

Protocol tools and runtimes can work at the RTPS layer without importing the DCPS entity model.

let (header, submessages) = parse_rtps_message(&packet)?;
for submessage in submessages {
    handle(header, submessage?);
}

retina-shm

Implicit same-host transport

writer.write(&value)

Per-reader transport selection. Eligible same-host readers receive a pool descriptor; remote readers receive RTPS DATA or DATA_FRAG.

One publish can fan out over shared memory and UDP with transport selected per reader.

writer.write(&sample)?;
// The same call serves local SHM and remote UDP readers.

Transport traits

Explicit transport composition

UdpStaticPeer::bind(config) · transport.send_submessage(bytes, &mut packet) · transport.receive_message(&mut packet)

Strategy-based transport composition. Static-peer routing and sync/async traits feed the same RTPS state machines.

Test harnesses and specialized runtimes can choose transport composition explicitly.

let config = StaticPeerConfig::new(local, remote, header);
let transport = UdpStaticPeer::bind(config)?;
transport.send_submessage(bytes, &mut packet)?;

Partial

DDS Security components

HandshakeSessionRegistry::new(limits) · decode_participant_generic_message(bytes, limits) · grant.authorize(request)

Bounded PKI-DH handshake state machine. Transcript verification, permissions decisions, and session transitions fail closed.

Security capability boundaries are available; complete secured data-plane composition and vendor interoperability are not claimed.

let mut sessions = HandshakeSessionRegistry::new(limits)?;
let message = decode_participant_generic_message(
    bytes, message_limits)?;

TypedReliableSession::new(transport, matched_participant, config)
session.write(value) -> Result<u64>
session.take_until(deadline) -> Result<T>
session.finish_until(deadline) -> Result<()>
session.metrics() -> ReliableMetrics