Building DDS applications with Retina

Tutorial 1 · about 20 minutes

Create typed DDS entities

A Retina Participant is a cloneable handle to one domain participant. Its shared state owns local topics, endpoint registrations, histories, matching state, QoS, and lifecycle. Writers and readers hold endpoint leases that keep the participant alive and unregister themselves when dropped.

What exists in memory

Participant handle
└─ Arc<Mutex<DomainParticipant>>
   ├─ topic and type descriptions
   ├─ writer and reader registrations
   ├─ local histories and matches
   └─ QoS, lifecycle, and status state

DataWriter<Temperature> ── endpoint lease ──┐
DataReader<Temperature> ── endpoint lease ──┴─ keep Participant alive

This first example uses one participant in one process. Tutorial 3 adds a RuntimeParticipant, discovery state, and UDP transport so another process can independently declare the same domain, topic, type, and QoS and then exchange RTPS data.

Define the type, topic, writer, and reader

use retina_core::{dds_type, Participant};

const TOPIC_NAME: &str = "sensors/temperature";
const EXPECTED_TEMPERATURE: f32 = 23.5;
const SAMPLE_EXPECTATION: &str = "one temperature sample";

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

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Join a running ROS 2 graph if ROS_DOMAIN_ID is set; otherwise use 0.
    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>(TOPIC_NAME);
    let writer = participant.create_writer(&topic)?;
    let reader = participant.create_reader(&topic)?;

    let expected = Temperature {
        degrees_celsius: EXPECTED_TEMPERATURE,
    };
    writer.write(&expected)?;
    let sample = reader.take()?.expect(SAMPLE_EXPECTATION);
    assert_eq!(sample.value, expected);
    Ok(())
}

What to verify

• The participant, writer, and reader use the same domain and topic name.
• The reader returns a typed Temperature, not an unstructured byte buffer.
• Changing the topic name or type identity prevents the endpoints from matching.

Tutorial 2 · about 15 minutes

Represent many robots as instances of one topic

A topic can carry many logical objects of the same type. This guide adds an instance key so one pose topic can represent many robots without creating a topic per robot. A bounded history and transient-local durability then define how much recent state the writer retains for each robot.

Add the key and retention policy

use retina_core::{
    DurabilityKind, HistoryKind, QosProfile, ReliabilityKind,
};

const TOPIC_NAME: &str = "fleet/pose";
const ROBOT_INSTANCE_KEY: &[u8] = b"robot-17";
const HISTORY_DEPTH: usize = 8;
const POSE_EXPECTATION: &str = "robot-17 pose";

let qos = QosProfile {
    reliability: ReliabilityKind::Reliable,
    durability: DurabilityKind::TransientLocal,
    history: HistoryKind::KeepLast,
    depth: HISTORY_DEPTH,
    ..QosProfile::default()
};

let topic = participant
    .topic::<Pose>(TOPIC_NAME)
    .with_qos(qos);
let writer = participant.create_writer(&topic)?;
let reader = participant.create_reader(&topic)?;

writer.write_instance(&pose, ROBOT_INSTANCE_KEY)?;
let pose = reader
    .take_instance(ROBOT_INSTANCE_KEY)?
    .expect(POSE_EXPECTATION);

What to verify

• Publishing another key on fleet/pose creates another logical instance, not another topic.
• take_instance() returns samples only for the requested robot key.
• KeepLast retains at most eight recent samples per instance, subject to the profile's resource limits.

Tutorial 3 · about 25 minutes

Run a reliable exchange over UDP

So far everything lived in one process. Now run two — a subscriber and a publisher in separate terminals. They find each other automatically (that's what SPDP and SEDP discovery do), then trade the same Temperature record from Tutorial 1 over reliable UDP, resending anything that gets dropped along the way.

Set the delivery contract

use retina_core::QosProfile;

const HISTORY_DEPTH: usize = 8;

let reliable_qos = QosProfile::builder()
    .reliable()
    .keep_last(HISTORY_DEPTH)
    .max_samples(HISTORY_DEPTH)
    .build();

// Apply the same compatible policy when creating both endpoints.
let topic = participant
    .topic::<Temperature>(TOPIC_NAME)
    .with_qos(reliable_qos);

The example peer below applies this contract internally and exposes the two process roles as command-line arguments.

Build the runtime in each process

use retina_runtime::participant::{
    DiscoveryRuntime, ParticipantIdentity, RuntimeParticipant,
};
use retina_runtime::reliable::{
    ReliableSessionConfig, TypedReliableSession,
};
use retina_runtime::{StaticPeerConfig, UdpStaticPeer};

let identity = ParticipantIdentity::new(
    DOMAIN_ID, participant_id, "temperature-peer");
let participant = RuntimeParticipant::<Temperature>::build(
    identity, writer_topic, reader_topic, reliable_qos)?;

let transport = UdpStaticPeer::bind(StaticPeerConfig::new(
    bind_address, peer_address, participant.header()))?;
let mut matched = participant.discover(
    &DiscoveryRuntime::default(), &transport, timeout)?;
let mut session = TypedReliableSession::new(
    &transport, &mut matched, ReliableSessionConfig::default());

Terminal 1 · subscriber

RETINA_MANIFEST="retina/Cargo.toml"
RETINA_PACKAGE="retina-tools"
TYPED_PEER="retina_typed_temperature_peer"
SUBSCRIBER_ADDRESS="127.0.0.1:56301"
PUBLISHER_ADDRESS="127.0.0.1:56300"

cargo run --manifest-path "${RETINA_MANIFEST}" \
  --package "${RETINA_PACKAGE}" \
  --bin "${TYPED_PEER}" -- \
  subscriber "${SUBSCRIBER_ADDRESS}" "${PUBLISHER_ADDRESS}"

Terminal 2 · publisher

RETINA_MANIFEST="retina/Cargo.toml"
RETINA_PACKAGE="retina-tools"
TYPED_PEER="retina_typed_temperature_peer"
PUBLISHER_ADDRESS="127.0.0.1:56300"
SUBSCRIBER_ADDRESS="127.0.0.1:56301"

cargo run --manifest-path "${RETINA_MANIFEST}" \
  --package "${RETINA_PACKAGE}" \
  --bin "${TYPED_PEER}" -- \
  publisher "${PUBLISHER_ADDRESS}" "${SUBSCRIBER_ADDRESS}"

TypedReliableSession handles the messy parts — buffers, serialization, and recovery. Samples too big for one packet are split into fragments automatically, and if a fragment or a whole sample goes missing, the reader asks for it and the writer resends. You don't write any of that retry logic yourself.

What to verify

• The independently started processes discover and match one writer with one reader.
• The subscriber receives the typed value and the publisher completes only after the reliable exchange.
• A best-effort writer would not satisfy a reader requesting reliable delivery.

Tutorial 4 · about 15 minutes

Use shared memory without changing publish code

You call the same write() from Tutorial 1. When a compatible reader is on the same machine and the shared-memory runtime is available, the runtime selects the local shared-memory path. A reader on another machine uses UDP. If both are present, each reader gets the transport appropriate to its location and capabilities.

Keep the application-level write unchanged

Continue with the typed entities and expected value from Tutorial 1:

const DELIVERY_EXPECTATION: &str = "at least one matched reader";

let delivered = writer.write(&expected)?;
assert!(delivered);

let sample = reader.take()?.expect(DELIVERY_EXPECTATION);
assert_eq!(sample.value, expected);

What to verify

• The writer and reader code is identical whether the selected data path is shared memory or UDP.
• An unavailable or incompatible shared-memory path falls back without changing the topic API.
• Reliability and type matching remain DDS concerns regardless of the selected transport.

Tutorial 5 · about 20 minutes

Bound a DDS Security handshake

Retina keeps handshake lifecycle separate from certificate and token cryptography. The registry below admits one discovered peer, correlates its reply with the original request, rejects replayed identities, and reaches the authenticated state only after the final message is sent.

use retina_security::{
    decode_participant_generic_message,
    GenericMessageLimits,
    HandshakeSessionLimits,
    HandshakeSessionRegistry,
    HandshakeSessionState,
};
use std::time::{Duration, Instant};

let mut sessions = HandshakeSessionRegistry::new(
    HandshakeSessionLimits {
        maximum_active_sessions: 8,
        maximum_token_bytes: 16 * 1024,
        maximum_retries: 3,
        timeout: Duration::from_secs(5),
    },
)?;

let started = Instant::now();
sessions.discover(peer_guid)?;
sessions.begin_request(
    peer_guid,
    request_identity,
    request_token.len(),
    started,
)?;

let reply = decode_participant_generic_message(
    &received_bytes,
    GenericMessageLimits {
        class_id_bytes: 256,
        message_tokens: 8,
        properties: 16,
        binary_properties: 16,
        property_name_bytes: 128,
        property_value_bytes: 4 * 1024,
        binary_value_bytes: 16 * 1024,
    },
)?;
sessions.receive_reply(
    peer_guid,
    reply.message_identity,
    reply.related_message_identity,
    encoded_token_bytes(&reply),
    Instant::now(),
)?;

verify_reply_signature_and_permissions(&reply)?;
assert!(sessions.mark_final_pending(peer_guid));
send_final_handshake_message(&reply)?;
assert!(sessions.confirm_final_delivery(peer_guid));
assert_eq!(
    sessions.state(peer_guid),
    Some(HandshakeSessionState::Authenticated),
);

What to verify

• A reply with the wrong related request identity is rejected.
• Repeating an accepted reply sequence is reported as a replay.
• Oversized tokens, expired deadlines, and exhausted retry budgets fail within configured limits.
• The session is not authenticated until cryptographic and permissions checks pass and the final message is delivered.

Tutorial 6 · about 20 minutes

Admit a teleoperation command

Create bounded command endpoints, grant one operator session control of one robot, and expose a command only after its robot, session, epoch, sequence, timestamp, and clock quality pass admission.

use retina_core::{dds_type, Participant};
use retina_teleop::{
    ClockQuality, CommandAdmission, CommandAdmissionConfig,
    CommandReadOutcome, ControlAuthority, ControlAuthorityConfig,
    DegradedClockPolicy, MonotonicTimestamp, MotionCommand,
    OperationMode, OperatorSessionId, RobotId, CommandSequence,
    SynchronizedTimestamp, TeleopEndpointFactory, TeleopProfile,
};

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

let robot = RobotId::new(17)?;
let session = OperatorSessionId::new(42)?;
let participant = Participant::new(7);
let factory = TeleopEndpointFactory::new(
    participant, TeleopProfile::default())?;

let mut writer = factory.create_motion_writer::<JointTarget>(
    "robot/motion")?;
let admission = CommandAdmission::new(CommandAdmissionConfig {
    robot,
    initial_authority: None,
    freshness_limit_nanoseconds: 5_000_000,
    degraded_clock_policy: DegradedClockPolicy::RequireSynchronized,
})
.ok_or("invalid admission policy")?;
let mut reader = factory.create_command_reader::<JointTarget>(
    "robot/motion", admission)?;

let mut authority = ControlAuthority::new(ControlAuthorityConfig {
    robot,
    lease_duration_ticks: 1_000,
})
.ok_or("invalid authority policy")?;
let grant = authority.grant(
    session, MonotonicTimestamp::from_ticks(10));
let lease = grant.lease.ok_or("authority not granted")?;
reader.adopt_authority(lease);

let command = MotionCommand {
    robot,
    session,
    epoch: lease.epoch,
    sequence: CommandSequence::new(1)?,
    generated_at: SynchronizedTimestamp::from_tai_nanoseconds(1_000_000),
    requested_mode: OperationMode::Enabled,
    payload: JointTarget { position: 0.25 },
};
writer.write(&command)?;

match reader.take(
    SynchronizedTimestamp::from_tai_nanoseconds(2_000_000),
    ClockQuality::Synchronized,
)? {
    CommandReadOutcome::Accepted(sample) => apply(sample.value.payload),
    CommandReadOutcome::Rejected(reason) => enter_safe_hold(reason),
    CommandReadOutcome::NoData => {}
}

What to verify

• A command is rejected until the reader adopts an active authority lease.
• A command from another session or an older control epoch is rejected.
• Repeating sequence 1 is reported as duplicate or out of order.
• A command outside the freshness limit never reaches apply().