Struct DataReader 

pub struct DataReader<T: DdsType> { /* private fields */ }

Typed DataReader — entnimmt Samples, die der RTPS-Reader fuer das Topic empfangen hat.

Live-Mode: rx: Some liefert Samples aus der Runtime-mpsc. Offline-Mode: in-memory inbox fuer Unit-Tests.

Implementations

impl<T: DdsType> DataReader<T>

pub fn with_filter<F>(self, filter: F) -> Self

where F: Fn(&T) -> bool + Send + Sync + 'static,

Setzt einen Content-Filter, der auf jedem Sample im take()- Pfad evaluiert wird. Rueckgabe false verwirft das Sample.

Builder-Stil: reader.with_filter(|s| s.value > 0).

.7a — SQL-Expression-Syntax via set_filter_expression folgt in .

pub fn topic(&self) -> &Topic<T>

Topic, von dem gelesen wird.

pub fn subscription_handle(&self) -> InstanceHandle

Spec §2.2.2.5.3.6 / §2.2.2.1.1 — InstanceHandle dieses DataReaders. Stabile Identitaet fuer DomainParticipant::contains_entity.

pub fn set_listener( &self, listener: Option<ArcDataReaderListener>, mask: StatusMask, )

setzt den DataReaderListener + StatusMask. None loescht den Slot. Spec §2.2.2.5.7.x set_listener.

pub fn get_listener(&self) -> Option<ArcDataReaderListener>

aktueller Listener-Klon, sofern vorhanden.

pub fn qos(&self) -> DataReaderQos

Aktuelle QoS (cloned, .1).

pub fn take(&self) -> Result<Vec<T>>

Nimmt alle zwischengespeicherten Samples und entfernt sie aus der Inbox. Liefert leeren Vec wenn nichts da ist.

Errors

• WireError wenn ein gespeicherter Payload sich nicht mehr decoden laesst (type-eval mismatch).

Examples found in repository

examples/hello_dds_subscriber.rs (line 40)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let factory = DomainParticipantFactory::instance();
    let participant = factory.create_participant(0, DomainParticipantQos::default())?;

    let topic = participant.create_topic::<RawBytes>("Chatter", TopicQos::default())?;
    let subscriber = participant.create_subscriber(SubscriberQos::default());
    let reader = subscriber.create_datareader::<RawBytes>(&topic, DataReaderQos::default())?;

    println!("hello_dds_subscriber: reading on Domain 0 Topic 'Chatter' — Ctrl-C to stop");

    loop {
        match reader.wait_for_data(Duration::from_secs(1)) {
            Ok(()) => {
                for sample in reader.take()? {
                    match std::str::from_utf8(&sample.data) {
                        Ok(s) => println!("  <- {s}"),
                        Err(_) => println!("  <- <{} bytes of non-UTF8>", sample.data.len()),
                    }
                }
            }
            Err(DdsError::Timeout) => {} // idle tick
            Err(e) => return Err(e.into()),
        }
    }
}

examples/shapes_demo_subscriber.rs (line 34)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let args: Vec<String> = env::args().collect();
    let topic_name = args.get(1).map_or("Square", String::as_str);
    let domain_id: i32 = args.get(2).and_then(|s| s.parse().ok()).unwrap_or(0);

    let factory = DomainParticipantFactory::instance();
    let participant = factory.create_participant(domain_id, DomainParticipantQos::default())?;
    let topic = participant.create_topic::<ShapeType>(topic_name, TopicQos::default())?;
    let subscriber = participant.create_subscriber(SubscriberQos::default());
    let reader = subscriber.create_datareader::<ShapeType>(&topic, DataReaderQos::default())?;

    println!("shapes_demo_subscriber: Topic={topic_name} Domain={domain_id} — Ctrl-C to stop");

    loop {
        match reader.wait_for_data(Duration::from_secs(1)) {
            Ok(()) => {
                for sample in reader.take()? {
                    println!(
                        "  <- color={:8} x={:4} y={:4} size={}",
                        sample.color, sample.x, sample.y, sample.shapesize
                    );
                }
            }
            Err(DdsError::Timeout) => {}
            Err(e) => return Err(e.into()),
        }
    }
}

examples/zerodds_perf.rs (line 105)

fn run_sub(runtime: Duration) -> Result<(), Box<dyn std::error::Error>> {
    let factory = DomainParticipantFactory::instance();
    let participant = factory.create_participant(0, DomainParticipantQos::default())?;
    let topic = participant.create_topic::<ShapeType>(PERF_TOPIC, TopicQos::default())?;
    let subscriber = participant.create_subscriber(SubscriberQos::default());
    let reader = subscriber.create_datareader::<ShapeType>(&topic, DataReaderQos::default())?;

    let total = Arc::new(AtomicU64::new(0));
    let start = Instant::now();
    let mut last_report = start;
    let mut last_total = 0u64;

    while start.elapsed() < runtime {
        if let Ok(samples) = reader.take() {
            for _ in samples {
                total.fetch_add(1, Ordering::Relaxed);
            }
        }
        if last_report.elapsed() >= Duration::from_secs(1) {
            let now_total = total.load(Ordering::Relaxed);
            let delta = now_total - last_total;
            let elapsed_s = last_report.elapsed().as_secs_f64();
            let rate_ks = (delta as f64 / elapsed_s) / 1000.0;
            println!(
                "{:.3}  size N total {} delta {} rate {:.2} kS/s",
                start.elapsed().as_secs_f64(),
                now_total,
                delta,
                rate_ks
            );
            last_report = Instant::now();
            last_total = now_total;
        }
        std::thread::sleep(Duration::from_millis(1));
    }
    println!(
        "# sub-done: total={} runtime={:.3}s",
        total.load(Ordering::Relaxed),
        start.elapsed().as_secs_f64()
    );
    Ok(())
}

fn run_pingpong(runtime: Duration) -> Result<(), Box<dyn std::error::Error>> {
    // Beide Rollen in einem Prozess via zwei Topics; Pinger schreibt
    // PerfPing (mit "send-time"-Marker als shapesize-i32-Diff), Ponger
    // (zweiter Prozess) liest und echot auf PerfPong. Dieser Prozess
    // ist der PINGER.
    let factory = DomainParticipantFactory::instance();
    let participant = factory.create_participant(0, DomainParticipantQos::default())?;
    let ping_topic = participant.create_topic::<ShapeType>(PING_TOPIC, TopicQos::default())?;
    let pong_topic = participant.create_topic::<ShapeType>(PONG_TOPIC, TopicQos::default())?;
    let publisher = participant.create_publisher(PublisherQos::default());
    let subscriber = participant.create_subscriber(SubscriberQos::default());
    let writer = publisher.create_datawriter::<ShapeType>(&ping_topic, DataWriterQos::default())?;
    let reader =
        subscriber.create_datareader::<ShapeType>(&pong_topic, DataReaderQos::default())?;

    let _ = writer.wait_for_matched_subscription(1, Duration::from_secs(5));
    let _ = reader.wait_for_matched_publication(1, Duration::from_secs(5));

    let mut rtts_us: Vec<u64> = Vec::new();
    let start = Instant::now();
    let mut seq = 0i32;
    while start.elapsed() < runtime {
        let send_us = now_micros();
        // shapesize codiert die unteren 32 bit von send_us als pseudo-id
        // (kollisionsanfaellig, aber nur fuer matching im Pong); echte
        // RTT-Berechnung passiert lokal via timestamp-storage.
        let _ = writer.write(&ShapeType::new("PING", 0, 0, seq));
        seq = seq.wrapping_add(1);

        // Wartet bis 50ms auf Pong
        let deadline = Instant::now() + Duration::from_millis(50);
        while Instant::now() < deadline {
            if let Ok(samples) = reader.take() {
                for s in samples {
                    if s.color == "PONG" && s.shapesize == seq.wrapping_sub(1) {
                        let recv_us = now_micros();
                        rtts_us.push(recv_us - send_us);
                        break;
                    }
                }
                if !rtts_us.is_empty() && rtts_us.last().is_some() {
                    break;
                }
            }
            std::thread::sleep(Duration::from_micros(100));
        }
        std::thread::sleep(Duration::from_millis(100)); // 10Hz ping rate
    }

    if rtts_us.is_empty() {
        println!("# pingpong: no RTTs collected — pong missing?");
        return Ok(());
    }
    rtts_us.sort_unstable();
    let n = rtts_us.len();
    let mean = rtts_us.iter().sum::<u64>() / n as u64;
    let p50 = rtts_us[n / 2];
    let p90 = rtts_us[(n * 9) / 10];
    let p99 = rtts_us[(n * 99) / 100];
    println!(
        "{:.3}  rtt mean {}us min {} 50% {} 90% {} 99% {} max {} cnt {}",
        start.elapsed().as_secs_f64(),
        mean,
        rtts_us.first().copied().unwrap_or(0),
        p50,
        p90,
        p99,
        rtts_us.last().copied().unwrap_or(0),
        n
    );
    Ok(())
}

fn run_pong(runtime: Duration) -> Result<(), Box<dyn std::error::Error>> {
    // Liest PerfPing, schreibt PerfPong mit gleichem shapesize.
    let factory = DomainParticipantFactory::instance();
    let participant = factory.create_participant(0, DomainParticipantQos::default())?;
    let ping_topic = participant.create_topic::<ShapeType>(PING_TOPIC, TopicQos::default())?;
    let pong_topic = participant.create_topic::<ShapeType>(PONG_TOPIC, TopicQos::default())?;
    let publisher = participant.create_publisher(PublisherQos::default());
    let subscriber = participant.create_subscriber(SubscriberQos::default());
    let reader =
        subscriber.create_datareader::<ShapeType>(&ping_topic, DataReaderQos::default())?;
    let writer = publisher.create_datawriter::<ShapeType>(&pong_topic, DataWriterQos::default())?;
    let _ = writer.wait_for_matched_subscription(1, Duration::from_secs(5));
    let _ = reader.wait_for_matched_publication(1, Duration::from_secs(5));

    let start = Instant::now();
    let mut echoed = 0u64;
    while start.elapsed() < runtime {
        if let Ok(samples) = reader.take() {
            for s in samples {
                if s.color == "PING" {
                    let _ = writer.write(&ShapeType::new("PONG", s.x, s.y, s.shapesize));
                    echoed += 1;
                }
            }
        }
        std::thread::sleep(Duration::from_micros(100));
    }
    println!(
        "# pong-done: echoed={echoed} runtime={:.3}s",
        start.elapsed().as_secs_f64()
    );
    Ok(())
}

examples/multi_endpoint_perf.rs (line 113)

fn run_sub_n(count: usize, runtime: Duration) -> Result<(), Box<dyn std::error::Error>> {
    let factory = DomainParticipantFactory::instance();
    let participant = factory.create_participant(0, DomainParticipantQos::default())?;
    let subscriber = participant.create_subscriber(SubscriberQos::default());

    let mut readers = Vec::with_capacity(count);
    println!("[multi_sub] creating {count} readers");
    let create_start = Instant::now();
    for i in 0..count {
        let topic = participant.create_topic::<ShapeType>(&topic_name(i), TopicQos::default())?;
        let reader = subscriber.create_datareader::<ShapeType>(&topic, DataReaderQos::default())?;
        readers.push(reader);
    }
    println!(
        "[multi_sub] created {count} readers in {:.2}s",
        create_start.elapsed().as_secs_f64()
    );

    let start = Instant::now();
    let mut total_received = 0u64;
    let mut last_report = start;
    while start.elapsed() < runtime {
        for r in &readers {
            if let Ok(samples) = r.take() {
                total_received += samples.len() as u64;
            }
        }
        if last_report.elapsed() >= Duration::from_secs(60) {
            println!(
                "{:.3}  multi_sub readers={} total_received={}",
                start.elapsed().as_secs_f64(),
                count,
                total_received
            );
            last_report = Instant::now();
        }
        std::thread::sleep(Duration::from_millis(10));
    }
    println!(
        "# multi_sub-done: readers={count} total_received={} runtime={:.3}s",
        total_received,
        start.elapsed().as_secs_f64()
    );
    Ok(())
}

examples/shapes_demo_viewer.rs (line 154)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let args: Vec<String> = env::args().collect();
    let domain_id: i32 = args.get(1).and_then(|s| s.parse().ok()).unwrap_or(0);

    let stop = Arc::new(AtomicBool::new(false));
    install_signal_handler(stop.clone());

    let factory = DomainParticipantFactory::instance();
    let participant = factory.create_participant(domain_id, DomainParticipantQos::default())?;
    let subscriber = participant.create_subscriber(SubscriberQos::default());

    let topics = ["Square", "Circle", "Triangle"];
    let mut readers = Vec::new();
    for t in &topics {
        let topic = participant.create_topic::<ShapeType>(t, TopicQos::default())?;
        let reader = subscriber.create_datareader::<ShapeType>(&topic, DataReaderQos::default())?;
        readers.push((*t, reader));
    }

    eprintln!(
        "shapes_demo_viewer: Domain={domain_id} — subscribed to Square / Circle / Triangle. Ctrl-C beendet."
    );
    eprintln!("Warte auf Discovery + erste Samples...");

    // Hide cursor + clear screen.
    print!("\x1B[?25l\x1B[2J");
    let mut stdout = std::io::stdout();
    stdout.flush().ok();

    // Pro (topic, color) speichern wir die letzte Position.
    let mut shapes: HashMap<(String, String), (i32, i32)> = HashMap::new();
    let mut sample_count: u64 = 0;
    let mut grid: Vec<Vec<(char, &'static str)>> = vec![vec![(' ', ""); VIEW_W]; VIEW_H];

    while !stop.load(Ordering::Relaxed) {
        // 1) Take all pending samples per topic.
        for (topic_name, reader) in &readers {
            if let Ok(samples) = reader.take() {
                for sample in samples {
                    shapes.insert(
                        ((*topic_name).to_string(), sample.color.clone()),
                        (sample.x, sample.y),
                    );
                    sample_count += 1;
                }
            }
        }

        // 2) Clear grid.
        for row in &mut grid {
            for cell in row {
                *cell = (' ', "");
            }
        }

        // 3) Plot shapes.
        for ((topic, color), (x, y)) in &shapes {
            let gx = map_x(*x);
            let gy = map_y(*y);
            if gy < VIEW_H && gx < VIEW_W {
                grid[gy][gx] = (glyph(topic), ansi_color(color));
            }
        }

        // 4) Render: home, draw, status line.
        print!("\x1B[H");
        // top border
        print!("┌");
        for _ in 0..VIEW_W {
            print!("─");
        }
        println!("┐");
        for row in &grid {
            print!("│");
            for (ch, color) in row {
                if color.is_empty() {
                    print!(" ");
                } else {
                    print!("{color}{ch}{ANSI_RESET}");
                }
            }
            println!("│");
        }
        print!("└");
        for _ in 0..VIEW_W {
            print!("─");
        }
        println!("┘");
        println!(
            "shapes={:3} samples={:6} domain={} — Ctrl-C beendet                            ",
            shapes.len(),
            sample_count,
            domain_id,
        );
        stdout.flush().ok();

        thread::sleep(Duration::from_millis(60));
    }

    // Show cursor again.
    print!("\x1B[?25h");
    println!("[shapes_demo_viewer] beendet. Total samples empfangen: {sample_count}");
    Ok(())
}

pub fn read(&self) -> Result<Vec<T>>

Liest alle Samples ohne sie zu entfernen. aktuell identisch zu take minus entfernen. Sample-State (ReadCondition §2.2.2.5.8) folgt im Wire-Up.

Errors

Wie take.

pub fn matched_publication_count(&self) -> usize

Anzahl matched Remote-Writer. Im Offline-Mode immer 0.

Spec: OMG DDS 1.4 §2.2.2.5.3.15 get_matched_publications.

Seiteneffekt — bei einer Aenderung des Matched-Count gegenueber dem letzten Aufruf wird on_subscription_matched via Bubble-Up-Kette gefeuert (Spec §2.2.4.2.6.7).

pub fn wait_for_matched_publication( &self, min_count: usize, timeout: Duration, ) -> Result<()>

Blockiert, bis mindestens min_count Remote-Writer matched sind oder timeout verstreicht. Event-driven via Runtime-Condvar (D.5e Phase-1) — wakup direkt wenn SEDP einen Match propagiert, kein 20-ms-Polling mehr.

Errors

DdsError::Timeout wenn min_count im Zeitfenster nicht erreicht wird.

Examples found in repository

examples/zerodds_perf.rs (line 151)

fn run_pingpong(runtime: Duration) -> Result<(), Box<dyn std::error::Error>> {
    // Beide Rollen in einem Prozess via zwei Topics; Pinger schreibt
    // PerfPing (mit "send-time"-Marker als shapesize-i32-Diff), Ponger
    // (zweiter Prozess) liest und echot auf PerfPong. Dieser Prozess
    // ist der PINGER.
    let factory = DomainParticipantFactory::instance();
    let participant = factory.create_participant(0, DomainParticipantQos::default())?;
    let ping_topic = participant.create_topic::<ShapeType>(PING_TOPIC, TopicQos::default())?;
    let pong_topic = participant.create_topic::<ShapeType>(PONG_TOPIC, TopicQos::default())?;
    let publisher = participant.create_publisher(PublisherQos::default());
    let subscriber = participant.create_subscriber(SubscriberQos::default());
    let writer = publisher.create_datawriter::<ShapeType>(&ping_topic, DataWriterQos::default())?;
    let reader =
        subscriber.create_datareader::<ShapeType>(&pong_topic, DataReaderQos::default())?;

    let _ = writer.wait_for_matched_subscription(1, Duration::from_secs(5));
    let _ = reader.wait_for_matched_publication(1, Duration::from_secs(5));

    let mut rtts_us: Vec<u64> = Vec::new();
    let start = Instant::now();
    let mut seq = 0i32;
    while start.elapsed() < runtime {
        let send_us = now_micros();
        // shapesize codiert die unteren 32 bit von send_us als pseudo-id
        // (kollisionsanfaellig, aber nur fuer matching im Pong); echte
        // RTT-Berechnung passiert lokal via timestamp-storage.
        let _ = writer.write(&ShapeType::new("PING", 0, 0, seq));
        seq = seq.wrapping_add(1);

        // Wartet bis 50ms auf Pong
        let deadline = Instant::now() + Duration::from_millis(50);
        while Instant::now() < deadline {
            if let Ok(samples) = reader.take() {
                for s in samples {
                    if s.color == "PONG" && s.shapesize == seq.wrapping_sub(1) {
                        let recv_us = now_micros();
                        rtts_us.push(recv_us - send_us);
                        break;
                    }
                }
                if !rtts_us.is_empty() && rtts_us.last().is_some() {
                    break;
                }
            }
            std::thread::sleep(Duration::from_micros(100));
        }
        std::thread::sleep(Duration::from_millis(100)); // 10Hz ping rate
    }

    if rtts_us.is_empty() {
        println!("# pingpong: no RTTs collected — pong missing?");
        return Ok(());
    }
    rtts_us.sort_unstable();
    let n = rtts_us.len();
    let mean = rtts_us.iter().sum::<u64>() / n as u64;
    let p50 = rtts_us[n / 2];
    let p90 = rtts_us[(n * 9) / 10];
    let p99 = rtts_us[(n * 99) / 100];
    println!(
        "{:.3}  rtt mean {}us min {} 50% {} 90% {} 99% {} max {} cnt {}",
        start.elapsed().as_secs_f64(),
        mean,
        rtts_us.first().copied().unwrap_or(0),
        p50,
        p90,
        p99,
        rtts_us.last().copied().unwrap_or(0),
        n
    );
    Ok(())
}

fn run_pong(runtime: Duration) -> Result<(), Box<dyn std::error::Error>> {
    // Liest PerfPing, schreibt PerfPong mit gleichem shapesize.
    let factory = DomainParticipantFactory::instance();
    let participant = factory.create_participant(0, DomainParticipantQos::default())?;
    let ping_topic = participant.create_topic::<ShapeType>(PING_TOPIC, TopicQos::default())?;
    let pong_topic = participant.create_topic::<ShapeType>(PONG_TOPIC, TopicQos::default())?;
    let publisher = participant.create_publisher(PublisherQos::default());
    let subscriber = participant.create_subscriber(SubscriberQos::default());
    let reader =
        subscriber.create_datareader::<ShapeType>(&ping_topic, DataReaderQos::default())?;
    let writer = publisher.create_datawriter::<ShapeType>(&pong_topic, DataWriterQos::default())?;
    let _ = writer.wait_for_matched_subscription(1, Duration::from_secs(5));
    let _ = reader.wait_for_matched_publication(1, Duration::from_secs(5));

    let start = Instant::now();
    let mut echoed = 0u64;
    while start.elapsed() < runtime {
        if let Ok(samples) = reader.take() {
            for s in samples {
                if s.color == "PING" {
                    let _ = writer.write(&ShapeType::new("PONG", s.x, s.y, s.shapesize));
                    echoed += 1;
                }
            }
        }
        std::thread::sleep(Duration::from_micros(100));
    }
    println!(
        "# pong-done: echoed={echoed} runtime={:.3}s",
        start.elapsed().as_secs_f64()
    );
    Ok(())
}

pub fn requested_deadline_missed_count(&self) -> u64

Counter fuer requested-Deadline-Verletzungen (Spec §2.2.4.2.11 REQUESTED_DEADLINE_MISSED_STATUS). Monoton steigend; steigt um 1 pro abgelaufenem Deadline-Fenster ohne empfangenes Sample. Offline / INFINITE → 0.

feuert ggf. on_requested_deadline_missed.

pub fn requested_incompatible_qos_status( &self, ) -> RequestedIncompatibleQosStatus

aktueller RequestedIncompatibleQosStatus. Spec §2.2.4.2.6.5. Triggert ggf. on_requested_incompatible_qos.

pub fn sample_lost_count(&self) -> u64

SampleLost-Counter. Spec §2.2.4.2.6.2.

pub fn sample_rejected_status(&self) -> SampleRejectedStatus

SampleRejected-Status. Spec §2.2.4.2.6.3.

pub fn drive_listeners(&self)

pollt alle Reader-Statuses einmal und feuert pending Listener. Convenience-Helper fuer Tests + periodische Tick-Aufrufer.

pub fn liveliness_changed_status(&self) -> (bool, u64, u64)

Liveliness-Status des matched Writers (Spec §2.2.4.2.14 LIVELINESS_CHANGED_STATUS): (alive, alive_count, not_alive_count).

• alive: aktueller Zustand (true = Writer hat Sample innerhalb seiner Lease-Duration geliefert).
• alive_count: Zaehler der “not_alive → alive”-Transitions.
• not_alive_count: Zaehler der “alive → not_alive”-Transitions.

Offline / INFINITE-Lease → (false, 0, 0) / (true, 0, 0) je nach Init. Fuer v1.3 wird nur LivelinessKind::Automatic ueberwacht.

pub fn wait_for_data(&self, timeout: Duration) -> Result<()>

Blockiert, bis mindestens ein Sample verfuegbar ist oder der Timeout abgelaufen ist. Das Sample wird dabei nicht entnommen — es wird in einen Staging-Buffer gelegt, den der naechste take() ausliest. Damit bleibt wait_for_data + take() der kanonische Subscriber-Loop, statt busy-polling im Application-Code.

Spec-Analog: OMG DDS 1.4 §2.2.2.5.8 ReadCondition + WaitSet. Diese API liefert die wichtigste Semantik (wake-on-data) ohne die komplette WaitSet/Condition-Infrastruktur.

Errors

DdsError::Timeout wenn im Zeitfenster nichts ankommt.

Examples found in repository

examples/hello_dds_subscriber.rs (line 38)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let factory = DomainParticipantFactory::instance();
    let participant = factory.create_participant(0, DomainParticipantQos::default())?;

    let topic = participant.create_topic::<RawBytes>("Chatter", TopicQos::default())?;
    let subscriber = participant.create_subscriber(SubscriberQos::default());
    let reader = subscriber.create_datareader::<RawBytes>(&topic, DataReaderQos::default())?;

    println!("hello_dds_subscriber: reading on Domain 0 Topic 'Chatter' — Ctrl-C to stop");

    loop {
        match reader.wait_for_data(Duration::from_secs(1)) {
            Ok(()) => {
                for sample in reader.take()? {
                    match std::str::from_utf8(&sample.data) {
                        Ok(s) => println!("  <- {s}"),
                        Err(_) => println!("  <- <{} bytes of non-UTF8>", sample.data.len()),
                    }
                }
            }
            Err(DdsError::Timeout) => {} // idle tick
            Err(e) => return Err(e.into()),
        }
    }
}

examples/shapes_demo_subscriber.rs (line 32)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let args: Vec<String> = env::args().collect();
    let topic_name = args.get(1).map_or("Square", String::as_str);
    let domain_id: i32 = args.get(2).and_then(|s| s.parse().ok()).unwrap_or(0);

    let factory = DomainParticipantFactory::instance();
    let participant = factory.create_participant(domain_id, DomainParticipantQos::default())?;
    let topic = participant.create_topic::<ShapeType>(topic_name, TopicQos::default())?;
    let subscriber = participant.create_subscriber(SubscriberQos::default());
    let reader = subscriber.create_datareader::<ShapeType>(&topic, DataReaderQos::default())?;

    println!("shapes_demo_subscriber: Topic={topic_name} Domain={domain_id} — Ctrl-C to stop");

    loop {
        match reader.wait_for_data(Duration::from_secs(1)) {
            Ok(()) => {
                for sample in reader.take()? {
                    println!(
                        "  <- color={:8} x={:4} y={:4} size={}",
                        sample.color, sample.x, sample.y, sample.shapesize
                    );
                }
            }
            Err(DdsError::Timeout) => {}
            Err(e) => return Err(e.into()),
        }
    }
}

pub fn instance_tracker(&self) -> InstanceTracker

Liefert den aktuellen InstanceTracker (geteilt mit der internen Buchhaltung). Hauptsaechlich fuer Tests / Inspection.

pub fn notify_writer_liveliness_lost(&self, writer_guid: [u8; 16]) -> usize

Spec §2.2.3.23 — Hook fuer “Writer X hat Liveliness verloren”. Macht zwei Dinge:

1. clear OWNERSHIP=EXCLUSIVE-Owner fuer alle Instanzen, deren Owner dieser Writer war (so dass der naechste Sample eines anderen Writers via should_accept_sample_under_exclusive_ownership neu gewinnen kann);
2. liefert die Anzahl betroffener Instanzen zurueck.

Wird aus dem WLP-Pfad gerufen, sobald ein Writer-Lease abgelaufen ist (siehe wlp::WlpEndpoint::lost_peers).

pub fn notify_participant_liveliness_lost(&self, prefix: [u8; 12]) -> usize

Wie Self::notify_writer_liveliness_lost, aber Match nur ueber die ersten 12 Bytes (GuidPrefix). Erlaubt Failover, wenn nur die Participant-Identitaet (z.B. bei SPDP-Lease-Expiry) bekannt ist.

pub fn lookup_instance(&self, instance: &T) -> InstanceHandle

Macht aus einem Sample-Wert den dazugehoerigen lokalen InstanceHandle, oder HANDLE_NIL wenn unbekannt / non-keyed. Spec §2.2.2.5.3.26 lookup_instance (Reader-Variante).

pub fn get_key_value(&self, handle: InstanceHandle) -> Result<T>

Spec §2.2.2.5.3.25 get_key_value. Liefert den Sample-Wert mit nur den @key-Feldern befuellt (rekonstruiert aus dem gespeicherten Key-Holder via T::decode).

Errors

BadParameter wenn handle unbekannt; WireError wenn T::decode den Key-Stream nicht rekonstruieren kann.

pub fn take_with_info(&self) -> Result<Vec<Sample<T>>>

take mit voller SampleInfo. Spec §2.2.2.5.3.5 take. Konsumiert die Samples aus dem Cache (NOT_READ → READ- Transition entfaellt, weil sie weg sind).

Errors

Wie Self::take.

pub fn read_with_info(&self) -> Result<Vec<Sample<T>>>

read mit voller SampleInfo. Konsumiert nicht — markiert die Samples nur als READ (Spec §2.2.2.5.3.4).

Errors

Wie Self::read.

pub fn take_filtered( &self, sample_mask: u32, view_mask: u32, instance_mask: u32, ) -> Result<Vec<Sample<T>>>

take mit State-Masken (Spec §2.2.2.5.3.6 take_w_condition).

Errors

Wie Self::take.

pub fn read_filtered( &self, sample_mask: u32, view_mask: u32, instance_mask: u32, ) -> Result<Vec<Sample<T>>>

read mit State-Masken (Spec §2.2.2.5.3.3 read_w_condition).

Errors

Wie Self::read.

pub fn read_w_condition( &self, condition: &Arc<QueryCondition>, ) -> Result<Vec<Sample<T>>>

read_w_condition (Spec §2.2.2.5.3.7) — wendet zusaetzlich zur State-Mask den SQL-Filter der QueryCondition pro Sample an. Samples bleiben im Cache (Sample-State NOT_READ → READ).

Errors

PreconditionNotMet bei Lock-Poisoning oder SQL-Eval-Fehler.

pub fn take_w_condition( &self, condition: &Arc<QueryCondition>, ) -> Result<Vec<Sample<T>>>

take_w_condition (Spec §2.2.2.5.3.8) — wie read_w_condition, aber konsumiert die Samples (entfernt aus dem Cache).

Errors

PreconditionNotMet bei Lock-Poisoning oder SQL-Eval-Fehler.

pub fn read_instance(&self, handle: InstanceHandle) -> Result<Vec<Sample<T>>>

read_instance (Spec §2.2.2.5.3.27). Liefert nur Samples der angegebenen Instanz.

Errors

BadParameter wenn handle == HANDLE_NIL.

pub fn take_instance(&self, handle: InstanceHandle) -> Result<Vec<Sample<T>>>

take_instance (Spec §2.2.2.5.3.27, Take-Variante). Konsumiert.

Errors

BadParameter wenn handle == HANDLE_NIL.

pub fn read_next_instance( &self, previous: InstanceHandle, ) -> Result<Vec<Sample<T>>>

read_next_instance (Spec §2.2.2.5.3.28). Liefert die Samples der naechsten Instanz (nach Sortier-Ordnung) hinter previous.

previous == HANDLE_NIL startet beim ersten Handle.

Errors

Wie read.

pub fn take_next_instance( &self, previous: InstanceHandle, ) -> Result<Vec<Sample<T>>>

take_next_instance (Spec §2.2.2.5.3.28). Take-Variante.

Errors

Wie take.

Trait Implementations

impl<T: DdsType> Debug for DataReader<T>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<T: DdsType> Entity for DataReader<T>

Available on crate feature std only.

fn set_qos(&self, qos: Self::Qos) -> Result<()>

Spec §2.2.3 / §2.2.2.5.3: DURABILITY, RELIABILITY, HISTORY, RESOURCE_LIMITS, OWNERSHIP sind Changeable=NO post-enable.

type Qos = DataReaderQos

QoS-Typ fuer diese Entity (z.B. DomainParticipantQos, DataWriterQos, …).

fn get_qos(&self) -> Self::Qos

Liefert die aktuelle QoS (clone). Spec §2.2.2.1.2 get_qos.

fn enable(&self) -> Result<()>

Enabled die Entity (idempotent). Spec §2.2.2.1.4 enable.

fn entity_state(&self) -> Arc<EntityState>

Interner Accessor — jede Impl liefert ihren Arc<EntityState>.

fn is_enabled(&self) -> bool

True wenn die Entity bereits enabled ist.

fn get_status_condition(&self) -> StatusCondition

StatusCondition dieser Entity. Spec §2.2.2.1.6 get_status_condition.

fn get_status_changes(&self) -> StatusMask

Bitmask der Status-Kinds, die seit letztem Read geaendert haben. Spec §2.2.2.1.5 get_status_changes.

fn get_instance_handle(&self) -> InstanceHandle

Lokaler 64-Bit-Identifier. Spec §2.2.2.1.7 get_instance_handle.