Struct Subscriber 

pub struct Subscriber { /* private fields */ }

Subscriber — Entity-Gruppe fuer DataReader.

Implementations

impl Subscriber

pub fn contains_reader(&self, handle: InstanceHandle) -> bool

Spec §2.2.2.2.1.10 — true wenn handle ein DataReader ist, der ueber diesen Subscriber erzeugt wurde.

pub fn begin_access(&self)

Spec §2.2.2.5.2.8 begin_access — markiert den Beginn eines kohaerenten Read-Sets. Verschachtelung ist erlaubt; jeder Aufruf erhoeht einen internen Counter, jedes end_access erniedrigt ihn.

pub fn end_access(&self) -> Result<()>

Spec §2.2.2.5.2.9 end_access — Gegenstueck zu begin_access.

Errors

DdsError::PreconditionNotMet wenn end_access ohne vorhergehendes begin_access gerufen wird.

pub fn is_access_open(&self) -> bool

true wenn aktuell ein Group-Access offen ist.

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

setzt den SubscriberListener + StatusMask. None loescht den Slot. Spec §2.2.2.5.6.x set_listener.

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

aktueller Listener-Klon.

pub fn create_datareader<T: DdsType + Send + 'static>( &self, topic: &Topic<T>, qos: DataReaderQos, ) -> Result<DataReader<T>>

Erzeugt einen typed DataReader<T>.

Errors

BadParameter bei Type-Name-Mismatch.

Examples found in repository

examples/hello_dds_subscriber.rs (line 33)

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 27)

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 97)

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 100)

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 132)

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(())
}

Trait Implementations

impl Debug for Subscriber

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

Formats the value using the given formatter.

impl Entity for Subscriber

Available on crate feature std only.

type Qos = SubscriberQos

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 set_qos(&self, qos: Self::Qos) -> Result<()>

Aendert QoS. Pre-enable: alles erlaubt. Post-enable: nur Felder mit “Changeable=YES” — sonst ImmutablePolicy-Error. Spec §2.2.2.1.2 set_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.