Struct DomainParticipant 

pub struct DomainParticipant { /* private fields */ }

Der Participant.

Implementations

impl DomainParticipant

pub fn runtime(&self) -> Option<&Arc<DcpsRuntime>>

Interner Zugriff auf die Runtime — von Publisher/Subscriber verwendet, um DataWriter/Reader anzulegen. None wenn der Participant im offline-Modus ist.

pub fn domain_id(&self) -> DomainId

Domain-Id.

pub fn qos(&self) -> DomainParticipantQos

Liefert eine Kopie der DomainParticipantQos (Spec §2.2.2.2.1.4 get_qos).

pub fn set_qos(&self, qos: DomainParticipantQos) -> Result<()>

Setzt die DomainParticipantQos (Spec §2.2.2.2.1.3 set_qos).

Errors

Aktuell keine — die Methode liefert Ok(()) immer. Spec laesst IMMUTABLE_POLICY zu, was wir aber nicht aktiv produzieren (alle Policies sind im RC1 mutable).

pub fn register_builtin_types(&self)

Registriert die 4 Spec-§7.6.5-Built-in-Types (DDS::String, DDS::KeyedString, DDS::Bytes, DDS::KeyedBytes) im lokalen TypeRegistry. Idempotent — doppelter Aufruf ueber- schreibt die Eintraege deterministisch.

Wird automatisch aus new()/new_with_runtime() aufgerufen, kann aber auch nach einem unregister_builtin_types()-Disable erneut aufgerufen werden.

pub fn unregister_builtin_types(&self)

Loescht alle registrierten Built-in-Types. Wird heute nicht von Default-Pfaden gerufen — Test-Hilfsfunktion fuer Disable-Flag-Tests.

pub fn find_builtin_type(&self, name: &str) -> Option<DynamicType>

Lookup eines Built-in-Types via Spec-Name (Spec §7.6.5). Gibt Some(DynamicType) zurueck wenn der Name bekannt ist (registriert via register_builtin_types).

pub fn registered_type_count(&self) -> usize

Anzahl registrierter Built-in-Types. Nach new() == 4.

pub fn enqueue_type_lookup(&self, hash: EquivalenceHash) -> bool

Versucht einen TypeLookup-Request fuer einen unbekannten EquivalenceHash zu queuen. Beachtet Backoff (5s zwischen Versuchen) und maximal 3 Wiederholungen pro Hash.

Returns: true wenn der Request gequeued wurde, false bei Backoff-Suppression oder Max-Attempts.

pub fn drain_type_lookup_requests(&self) -> Vec<(EquivalenceHash, u64)>

Drainet die queued TypeLookup-Requests. Liefert Vec<(hash, seq)>. In Production-Umgebung wuerde der Caller die Hashes via TypeLookupClient + Reliable-Writer auf den TL_SVC_REQ_WRITER-Endpoint senden.

pub fn ingest_type_lookup_reply( &self, types: Vec<(EquivalenceHash, MinimalTypeObject)>, ) -> usize

Empfaengt ein TypeLookup-Reply (TypeObjects pro Hash). Registriert die TypeObjects in einem internen TypeRegistry- Spiegel — danach kann ein gestoppter QoS-Match retried werden.

Anzahl erfolgreich registrierter Typen wird zurueckgegeben.

pub fn on_remote_publication_discovered( &self, type_information_blob: Option<&[u8]>, ) -> usize

SEDP-Discovery-Hook: prueft eine eingehende PublicationBuiltinTopicData auf Type-Hashes, die lokal nicht aufloesbar sind. Bei Bedarf wird ein TypeLookup-Request via enqueue_type_lookup gequeued.

Der RPC-Pfad ist via DcpsRuntime::send_type_lookup_request auf den TL_SVC_REQ_*-Endpoints (XTypes 1.3 §7.6.3.3.4) live; diese Methode entscheidet pro Hash, ob ein Re-Request lohnt (lokale Registry-Lookup + Backoff-Tracking).

Returns: Anzahl gequeued unbekannter Hashes (max 2 — minimal + complete).

pub fn on_remote_subscription_discovered( &self, type_information_blob: Option<&[u8]>, ) -> usize

SEDP-Discovery-Hook fuer SubscriptionBuiltinTopicData. Symmetrisch zu on_remote_publication_discovered.

pub fn type_lookup_exhausted(&self, hash: EquivalenceHash) -> bool

True wenn fuer den Hash bereits MAX_ATTEMPTS erreicht. Wird vom Match-Re-Try-Pfad konsultiert: spaeter aufgeben statt endlos zu pollen.

pub fn create_topic<T: DdsType>( &self, name: &str, qos: TopicQos, ) -> Result<Topic<T>>

Erzeugt einen typed Topic-Handle. Wiederholte Aufrufe mit gleichem Namen + Typ liefern denselben Handle (Ref-geteilt).

Errors

• InconsistentPolicy wenn ein Topic mit diesem Namen bereits unter anderem Typ registriert ist.
• BadParameter bei leerem Namen.

Examples found in repository

examples/hello_dds_subscriber.rs (line 31)

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/hello_dds_publisher.rs (line 42)

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 publisher = participant.create_publisher(PublisherQos::default());
    let writer = publisher.create_datawriter::<RawBytes>(&topic, DataWriterQos::default())?;

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

    // Ein paar Sekunden warten, damit SPDP+SEDP Subscriber entdecken kann.
    // Ohne Warte-Phase gehen die ersten Samples ins Leere (noch kein Reader
    // gematched). v1.3 bringt `wait_for_matched_subscription`.
    thread::sleep(Duration::from_secs(3));

    let mut counter: u32 = 0;
    loop {
        let payload = format!("hello #{counter}");
        writer.write(&RawBytes::new(payload.clone().into_bytes()))?;
        println!("  -> {payload}");
        counter = counter.wrapping_add(1);
        thread::sleep(Duration::from_secs(1));
    }
}

examples/shapes_demo_subscriber.rs (line 25)

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

fn run_pub(size: usize, 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 publisher = participant.create_publisher(PublisherQos::default());
    let writer = publisher.create_datawriter::<ShapeType>(&topic, DataWriterQos::default())?;

    // Discovery-Wait
    let _ = writer.wait_for_matched_subscription(1, Duration::from_secs(5));

    let payload_color = "X".repeat(size.saturating_sub(4 * 4)); // size minus i32-Felder approx
    let start = Instant::now();
    let mut total = 0u64;
    let mut last_report = start;
    let mut samples_since_report = 0u64;

    while start.elapsed() < runtime {
        let s = ShapeType::new(payload_color.clone(), 0, 0, total as i32);
        if writer.write(&s).is_ok() {
            total += 1;
            samples_since_report += 1;
        }
        // Report jede Sekunde
        if last_report.elapsed() >= Duration::from_secs(1) {
            let elapsed_s = last_report.elapsed().as_secs_f64();
            let rate_ks = (samples_since_report as f64 / elapsed_s) / 1000.0;
            let rate_mb =
                (samples_since_report as f64 * size as f64 * 8.0 / elapsed_s) / 1_000_000.0;
            println!(
                "{:.3}  size {} total {} rate {:.2} kS/s {:.2} Mb/s",
                start.elapsed().as_secs_f64(),
                size,
                total,
                rate_ks,
                rate_mb
            );
            last_report = Instant::now();
            samples_since_report = 0;
        }
    }
    println!(
        "# pub-done: total={total} runtime={:.3}s",
        start.elapsed().as_secs_f64()
    );
    Ok(())
}

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/shapes_demo_publisher.rs (line 42)

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 color = args.get(2).map_or("BLUE", String::as_str);
    let domain_id: i32 = args.get(3).and_then(|s| s.parse().ok()).unwrap_or(0);

    if !["Square", "Circle", "Triangle"].contains(&topic_name) {
        eprintln!(
            "warning: topic '{topic_name}' ist kein Standard-ShapesDemo-Topic (Square/Circle/Triangle)",
        );
    }

    let factory = DomainParticipantFactory::instance();
    let participant = factory.create_participant(domain_id, DomainParticipantQos::default())?;
    let topic = participant.create_topic::<ShapeType>(topic_name, TopicQos::default())?;
    let publisher = participant.create_publisher(PublisherQos::default());
    let writer = publisher.create_datawriter::<ShapeType>(&topic, DataWriterQos::default())?;

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

    // Discovery-Wait — bis mindestens 1 Subscriber gematched hat.
    let matched = writer.wait_for_matched_subscription(1, Duration::from_secs(10));
    match matched {
        Ok(()) => println!("matched subscriber found, starting publication"),
        Err(_) => println!("no subscriber in 10s — publishing anyway, samples may drop"),
    }

    // Sinus-Bewegung auf 240x270-Canvas (ShapesDemo-Default).
    let shapesize = 30;
    let mut t: f32 = 0.0;
    loop {
        let x = (120.0 + 80.0 * (t).sin()) as i32;
        let y = (135.0 + 90.0 * (t * 1.3).cos()) as i32;
        let sample = ShapeType::new(color, x, y, shapesize);
        writer.write(&sample)?;
        println!("  -> color={color} x={x} y={y} size={shapesize}");
        t += 0.15;
        thread::sleep(Duration::from_millis(100));
    }
}

examples/multi_endpoint_perf.rs (line 45)

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

    let mut writers = Vec::with_capacity(count);
    println!("[multi_pub] creating {count} writers");
    let create_start = Instant::now();
    for i in 0..count {
        let topic = participant.create_topic::<ShapeType>(&topic_name(i), TopicQos::default())?;
        let writer = publisher.create_datawriter::<ShapeType>(&topic, DataWriterQos::default())?;
        writers.push(writer);
    }
    println!(
        "[multi_pub] created {count} writers in {:.2}s",
        create_start.elapsed().as_secs_f64()
    );

    let start = Instant::now();
    let mut tick = 0u64;
    let mut total_writes = 0u64;
    while start.elapsed() < runtime {
        let tick_start = Instant::now();
        for (i, w) in writers.iter().enumerate() {
            let s = ShapeType::new(format!("MEP{i:04}"), tick as i32, 0, 30);
            if w.write(&s).is_ok() {
                total_writes += 1;
            }
        }
        tick += 1;
        // 1 Hz pro Writer — schlaeft bis zur naechsten Sekunde
        if tick % 60 == 0 {
            println!(
                "{:.3}  multi_pub tick={} writers={} total_writes={} loop_us={}",
                start.elapsed().as_secs_f64(),
                tick,
                count,
                total_writes,
                tick_start.elapsed().as_micros()
            );
        }
        let elapsed = tick_start.elapsed();
        if elapsed < Duration::from_secs(1) {
            std::thread::sleep(Duration::from_secs(1) - elapsed);
        }
    }
    println!(
        "# multi_pub-done: writers={count} total_writes={} runtime={:.3}s",
        total_writes,
        start.elapsed().as_secs_f64()
    );
    Ok(())
}

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

Additional examples can be found in:

• examples/shapes_demo_viewer.rs

pub fn lookup_topicdescription( &self, name: &str, ) -> Option<TopicDescriptionHandle>

Sofortiger lokaler Lookup eines Topics nach Name — gibt None zurueck, wenn kein lokales create_topic mit diesem Namen erfolgt ist. Macht keinen Discovery-Wait (das ist find_topic). Spec-Referenz: OMG DDS 1.4 §2.2.2.2.1.12 “lookup_topicdescription”.

pub fn find_topic( &self, name: &str, timeout: Duration, ) -> Result<TopicDescriptionHandle>

Wartet bis ein Topic mit dem gegebenen Namen via Discovery (SEDP-Publication oder -Subscription) sichtbar ist — oder bis timeout abgelaufen ist. Spec-Referenz: OMG DDS 1.4 §2.2.2.2.1.11 find_topic.

Returns:

• Ok(handle) mit Name + Type-Name + Participant, falls waehrend timeout ein passendes SEDP-Endpoint sichtbar wurde. Lokale Topics zaehlen ebenfalls (keine Pflicht zu warten, wenn create_topic schon lief).
• Err(Timeout) wenn timeout abgelaufen ist.

Errors

• DdsError::Timeout wenn timeout ohne Discovery-Match abgelaufen ist.
• DdsError::BadParameter bei leerem Namen.

pub fn create_contentfilteredtopic<T: DdsType>( &self, name: &str, related_topic: &Topic<T>, filter_expression: &str, filter_parameters: Vec<String>, ) -> Result<ContentFilteredTopic<T>>

Erzeugt ein ContentFilteredTopic als Subset eines bereits vorhandenen Topic<T>. Spec-Referenz: OMG DDS 1.4 §2.2.2.2.1.13 create_contentfilteredtopic.

Die filter_expression ist ein SQL-Subset (siehe Annex B). filter_parameters sind Strings, die %0, %1, … in der Expression ersetzen.

Errors

• BadParameter bei leerem Namen oder leerer Expression.
• BadParameter wenn die Filter-Expression nicht parst.
• BadParameter wenn ein referenzierter %N-Parameter nicht im filter_parameters-Vec geliefert wird.

pub fn create_multitopic<T: DdsType>( &self, name: &str, type_name: &str, related_topic_names: Vec<String>, subscription_expression: &str, expression_parameters: Vec<String>, ) -> Result<MultiTopic<T>>

Erzeugt eine MultiTopic als kombinierende TopicDescription ueber 1+ Underlying-Topics mit SQL-Subscription-Expression. Spec-Referenz: OMG DDS 1.4 §2.2.2.2.1.15 create_multitopic (optionales Spec-Feature).

Errors

• BadParameter bei leerem Namen oder Type-Namen.
• BadParameter wenn related_topic_names leer ist.
• BadParameter wenn die Subscription-Expression nicht parst.
• BadParameter wenn ein referenzierter %N-Parameter nicht im expression_parameters-Vec geliefert wird.

pub fn delete_multitopic<T: DdsType>(&self, mt: &MultiTopic<T>) -> Result<()>

Loescht eine MultiTopic. Spec §2.2.2.2.1.16 delete_multitopic. v1.2 ist es ein no-op-shim mit Participant- Match-Check.

Errors

BadParameter wenn die MultiTopic zu einem anderen Participant gehoert.

pub fn delete_contentfilteredtopic<T: DdsType>( &self, cft: &ContentFilteredTopic<T>, ) -> Result<()>

Loescht ein ContentFilteredTopic. Spec-Referenz: §2.2.2.2.1.14 delete_contentfilteredtopic.

In Rust ist das Lifetime-Handle des CFT bereits durch Drop abgedeckt — die zugrundeliegenden Ressourcen werden frei, sobald der ContentFilteredTopic<T> aus dem Scope geht. Diese Methode existiert fuer Spec-Compliance der C++-API und validiert den Participant-Match (Spec verlangt BadParameter, wenn das CFT zu einem anderen Participant gehoert).

Errors

• BadParameter wenn das CFT zu einem anderen Participant gehoert.

pub fn create_publisher(&self, qos: PublisherQos) -> Publisher

Erzeugt einen Publisher mit gegebener QoS (Default reicht fuer v1.2).

Examples found in repository

examples/hello_dds_publisher.rs (line 43)

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 publisher = participant.create_publisher(PublisherQos::default());
    let writer = publisher.create_datawriter::<RawBytes>(&topic, DataWriterQos::default())?;

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

    // Ein paar Sekunden warten, damit SPDP+SEDP Subscriber entdecken kann.
    // Ohne Warte-Phase gehen die ersten Samples ins Leere (noch kein Reader
    // gematched). v1.3 bringt `wait_for_matched_subscription`.
    thread::sleep(Duration::from_secs(3));

    let mut counter: u32 = 0;
    loop {
        let payload = format!("hello #{counter}");
        writer.write(&RawBytes::new(payload.clone().into_bytes()))?;
        println!("  -> {payload}");
        counter = counter.wrapping_add(1);
        thread::sleep(Duration::from_secs(1));
    }
}

examples/zerodds_perf.rs (line 49)

fn run_pub(size: usize, 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 publisher = participant.create_publisher(PublisherQos::default());
    let writer = publisher.create_datawriter::<ShapeType>(&topic, DataWriterQos::default())?;

    // Discovery-Wait
    let _ = writer.wait_for_matched_subscription(1, Duration::from_secs(5));

    let payload_color = "X".repeat(size.saturating_sub(4 * 4)); // size minus i32-Felder approx
    let start = Instant::now();
    let mut total = 0u64;
    let mut last_report = start;
    let mut samples_since_report = 0u64;

    while start.elapsed() < runtime {
        let s = ShapeType::new(payload_color.clone(), 0, 0, total as i32);
        if writer.write(&s).is_ok() {
            total += 1;
            samples_since_report += 1;
        }
        // Report jede Sekunde
        if last_report.elapsed() >= Duration::from_secs(1) {
            let elapsed_s = last_report.elapsed().as_secs_f64();
            let rate_ks = (samples_since_report as f64 / elapsed_s) / 1000.0;
            let rate_mb =
                (samples_since_report as f64 * size as f64 * 8.0 / elapsed_s) / 1_000_000.0;
            println!(
                "{:.3}  size {} total {} rate {:.2} kS/s {:.2} Mb/s",
                start.elapsed().as_secs_f64(),
                size,
                total,
                rate_ks,
                rate_mb
            );
            last_report = Instant::now();
            samples_since_report = 0;
        }
    }
    println!(
        "# pub-done: total={total} runtime={:.3}s",
        start.elapsed().as_secs_f64()
    );
    Ok(())
}

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/shapes_demo_publisher.rs (line 43)

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 color = args.get(2).map_or("BLUE", String::as_str);
    let domain_id: i32 = args.get(3).and_then(|s| s.parse().ok()).unwrap_or(0);

    if !["Square", "Circle", "Triangle"].contains(&topic_name) {
        eprintln!(
            "warning: topic '{topic_name}' ist kein Standard-ShapesDemo-Topic (Square/Circle/Triangle)",
        );
    }

    let factory = DomainParticipantFactory::instance();
    let participant = factory.create_participant(domain_id, DomainParticipantQos::default())?;
    let topic = participant.create_topic::<ShapeType>(topic_name, TopicQos::default())?;
    let publisher = participant.create_publisher(PublisherQos::default());
    let writer = publisher.create_datawriter::<ShapeType>(&topic, DataWriterQos::default())?;

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

    // Discovery-Wait — bis mindestens 1 Subscriber gematched hat.
    let matched = writer.wait_for_matched_subscription(1, Duration::from_secs(10));
    match matched {
        Ok(()) => println!("matched subscriber found, starting publication"),
        Err(_) => println!("no subscriber in 10s — publishing anyway, samples may drop"),
    }

    // Sinus-Bewegung auf 240x270-Canvas (ShapesDemo-Default).
    let shapesize = 30;
    let mut t: f32 = 0.0;
    loop {
        let x = (120.0 + 80.0 * (t).sin()) as i32;
        let y = (135.0 + 90.0 * (t * 1.3).cos()) as i32;
        let sample = ShapeType::new(color, x, y, shapesize);
        writer.write(&sample)?;
        println!("  -> color={color} x={x} y={y} size={shapesize}");
        t += 0.15;
        thread::sleep(Duration::from_millis(100));
    }
}

examples/multi_endpoint_perf.rs (line 39)

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

    let mut writers = Vec::with_capacity(count);
    println!("[multi_pub] creating {count} writers");
    let create_start = Instant::now();
    for i in 0..count {
        let topic = participant.create_topic::<ShapeType>(&topic_name(i), TopicQos::default())?;
        let writer = publisher.create_datawriter::<ShapeType>(&topic, DataWriterQos::default())?;
        writers.push(writer);
    }
    println!(
        "[multi_pub] created {count} writers in {:.2}s",
        create_start.elapsed().as_secs_f64()
    );

    let start = Instant::now();
    let mut tick = 0u64;
    let mut total_writes = 0u64;
    while start.elapsed() < runtime {
        let tick_start = Instant::now();
        for (i, w) in writers.iter().enumerate() {
            let s = ShapeType::new(format!("MEP{i:04}"), tick as i32, 0, 30);
            if w.write(&s).is_ok() {
                total_writes += 1;
            }
        }
        tick += 1;
        // 1 Hz pro Writer — schlaeft bis zur naechsten Sekunde
        if tick % 60 == 0 {
            println!(
                "{:.3}  multi_pub tick={} writers={} total_writes={} loop_us={}",
                start.elapsed().as_secs_f64(),
                tick,
                count,
                total_writes,
                tick_start.elapsed().as_micros()
            );
        }
        let elapsed = tick_start.elapsed();
        if elapsed < Duration::from_secs(1) {
            std::thread::sleep(Duration::from_secs(1) - elapsed);
        }
    }
    println!(
        "# multi_pub-done: writers={count} total_writes={} runtime={:.3}s",
        total_writes,
        start.elapsed().as_secs_f64()
    );
    Ok(())
}

pub fn create_subscriber(&self, qos: SubscriberQos) -> Subscriber

Erzeugt einen Subscriber.

Examples found in repository

examples/hello_dds_subscriber.rs (line 32)

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

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

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

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

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 topics_len(&self) -> usize

Anzahl aktuell registrierter Topics. Diagnose-API.

pub fn discovered_participants_count(&self) -> usize

Anzahl aktuell entdeckter Remote-Participants ueber SPDP. Spec: OMG DDS 1.4 §2.2.2.2.1.7 get_discovered_participants. 0 im offline-Modus.

pub fn discovered_publications_count(&self) -> usize

Anzahl aktuell im SEDP-Cache bekannter Remote-Publications. Spec: OMG DDS 1.4 §2.2.2.2.1.9 get_discovered_topics (~analog).

pub fn discovered_subscriptions_count(&self) -> usize

Anzahl aktuell im SEDP-Cache bekannter Remote-Subscriptions.

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

Markiert einen entdeckten remote DomainParticipant als “ignoriert” — alle weiteren SPDP-Beacons mit diesem Handle fallen aus dem Builtin-Reader-Stream raus, und gleichzeitig werden alle SEDP-Endpoints, die zum gleichen Participant- Prefix gehoeren, ebenfalls verworfen (Spec §2.2.2.2.1.14).

Per Spec ist die Aktion monoton — ein einmal ignorierter Participant bleibt es fuer den Lebenszyklus dieses Participants.

Errors

Aktuell keine — die Methode liefert Ok(()) immer. Spec laesst OUT_OF_RESOURCES zu, was wir aber nicht aktiv produzieren.

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

Markiert ein entdecktes remote Topic als “ignoriert”. Spec §2.2.2.2.1.15.

Errors

Wie Self::ignore_participant.

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

Markiert eine entdeckte remote Publication als “ignoriert”. Spec §2.2.2.2.1.16.

Errors

Wie Self::ignore_participant.

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

Markiert eine entdeckte remote Subscription als “ignoriert”. Spec §2.2.2.2.1.17.

Errors

Wie Self::ignore_participant.

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

true wenn handle per ignore_participant markiert wurde.

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

true wenn handle per ignore_topic markiert wurde.

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

true wenn handle per ignore_publication markiert wurde.

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

true wenn handle per ignore_subscription markiert wurde.

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

Loescht alle vom Participant gehaltenen Children (Publishers, Subscribers, Topics, Builtin-Reader-Inboxes). Spec §2.2.2.2.1.18 — analoger Pendant existiert in Publisher/Subscriber/DataReader, der hier rekursiv mit abgedeckt wird.

Offline-Verhalten:

• Topic-Registry geleert (lokale Topics).
• Publisher-/Subscriber-Tracker geleert.
• Builtin-Topic-Reader-Inboxes geleert (so dass take() nach delete_contained_entities ein leeres Vec liefert).
• Kein SEDP-Unannounce — das Live-Verhalten uebernimmt das, sobald die Runtime ein Drop/shutdown-Handle bekommt. Aktueller Stand: der Runtime-Thread laeuft bis zum Process-Exit.

Errors

PreconditionNotMet wenn ein interner Mutex vergiftet ist.

pub fn publishers_len(&self) -> usize

Anzahl der per create_publisher getrackten Publisher. Diagnose-API fuer Tests.

pub fn subscribers_len(&self) -> usize

Anzahl der per create_subscriber getrackten Subscriber.

pub fn instance_handle(&self) -> InstanceHandle

Liefert den InstanceHandle dieses Participants. Identifiziert die Entity gegenueber DCPS-API-Konsumenten (Spec §2.2.2.1.1 get_instance_handle).

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

Spec §2.2.2.2.1.10 contains_entity — true wenn handle zu diesem Participant oder einer seiner direkt oder rekursiv enthaltenen Entities gehoert.

Eingeschlossene Entity-Typen:

• der Participant selbst
• alle per create_topic registrierten Topics
• alle per create_publisher / create_subscriber erzeugten Publisher/Subscriber
• rekursiv: alle per Publisher::create_datawriter / Subscriber::create_datareader erzeugten DataWriter/DataReader.

pub fn get_discovered_participants(&self) -> Vec<InstanceHandle>

Liefert die InstanceHandles aller aktuell entdeckten remote Participants (Spec §2.2.2.2.1.27). Im offline-Modus leer. Ignorierte Participants tauchen nicht auf.

pub fn get_discovered_participant_data( &self, handle: InstanceHandle, ) -> Result<ParticipantBuiltinTopicData>

Liefert die ParticipantBuiltinTopicData zu einem Handle aus get_discovered_participants (Spec §2.2.2.2.1.28).

Errors

BadParameter wenn handle keinen entdeckten Participant referenziert (oder wenn er ignoriert wurde).

pub fn get_discovered_topics(&self) -> Vec<InstanceHandle>

Liefert die InstanceHandles aller aktuell entdeckten remote Topics. Spec §2.2.2.2.1.29.

Topics werden via SEDP-Pub/Sub-Announcements indirekt entdeckt — pro (topic_name, type_name) synthetisieren wir einen stabilen Schluessel via TopicBuiltinTopicData:: synthesize_key. Ignorierte Topics tauchen nicht auf.

pub fn get_discovered_topic_data( &self, handle: InstanceHandle, ) -> Result<TopicBuiltinTopicData>

Liefert die TopicBuiltinTopicData zu einem Handle aus get_discovered_topics. Spec §2.2.2.2.1.30.

Errors

BadParameter wenn handle keinem entdeckten Topic entspricht (oder ignoriert wurde).

pub fn get_builtin_subscriber(&self) -> Arc<BuiltinSubscriber>

Builtin-Subscriber des Participants (DDS 1.4 §2.2.2.2.1.7).

Liefert immer denselben Subscriber-Handle (genau ein Builtin-Subscriber pro Participant). Er enthaelt 4 vor-erzeugte Reader fuer die Builtin-Topics:

• DCPSParticipant → ParticipantBuiltinTopicData
• DCPSTopic → TopicBuiltinTopicData
• DCPSPublication → PublicationBuiltinTopicData
• DCPSSubscription → SubscriptionBuiltinTopicData

SPDP-/SEDP-Receive triggert intern einen Sample-Insert, der per take()/read() abgeholt werden kann (DDS 1.4 §2.2.5).

Example

use zerodds_dcps::*;
let participant = DomainParticipantFactory::instance()
    .create_participant_offline(0, DomainParticipantQos::default());
let bs = participant.get_builtin_subscriber();
let r = bs
    .lookup_datareader::<DcpsParticipantBuiltinTopicData>("DCPSParticipant")
    .expect("builtin reader");
// Anfangs leer (offline-Mode → keine SPDP-Empfange).
assert!(r.take().expect("take").is_empty());

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

Setzt den DomainParticipantListener. listener=None loescht den Slot. mask ist die StatusMask, die festlegt, welche Status-Bits dieser Listener konsumiert (Spec §2.2.4.2.3 Bubble-Up).

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

Liefert den aktuell installierten Listener-Klon, falls vorhanden. Spec §2.2.2.2.3.x get_listener.

Trait Implementations

impl Clone for DomainParticipant

fn clone(&self) -> DomainParticipant

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for DomainParticipant

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

Formats the value using the given formatter.

impl Entity for DomainParticipant

type Qos = DomainParticipantQos

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.