Struct DataWriter 

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

Typed DataWriter — schickt Samples an alle matched Reader des Topics.

Zwei Modi:

• Live (runtime: Some, entity_id: Some): write() delegiert an die Runtime → ReliableWriter → UDP.
• Offline (Offline-Fallback, Runtime=None): write() queued in-memory; fuer Unit-Tests ohne Netz.

Implementations

impl<T: DdsType> DataWriter<T>

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

Topic, an das gesendet wird.

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

setzt den DataWriterListener + StatusMask. None loescht den Slot. Spec §2.2.2.4.2.x set_listener.

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

aktueller Listener-Klon, sofern vorhanden.

pub fn qos(&self) -> DataWriterQos

Aktuelle QoS (cloned, .1).

pub fn write(&self, sample: &T) -> Result<()>

Sendet einen Sample an alle matched Reader.

Spec §2.2.3.19 RESOURCE_LIMITS Reliable-Block: Wenn der lokale Writer-Cache max_samples erreicht hat UND Reliability=RELIABLE UND max_blocking_time > 0, blockt write() bis Reader ACK den Slot freigibt oder das Timeout abgelaufen ist. Im Best-Effort-Mode oder mit max_blocking_time = 0 schlaegt write() sofort fehl mit OutOfResources.

Errors

• WireError wenn T::encode fehlschlaegt.
• OutOfResources wenn die Queue full ist + Best-Effort/keine Block-Zeit, oder wenn der Block-Timeout vor einem Drain abgelaufen ist.
• PreconditionNotMet bei Lock-Poisoning.

Examples found in repository

examples/hello_dds_publisher.rs (line 56)

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

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

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

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

Anzahl bisher geschriebener Samples. Test-Hilfsfunktion, in Runtime durch echte HistoryCache-Counter ersetzt.

pub fn matched_subscription_count(&self) -> usize

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

Spec: OMG DDS 1.4 §2.2.2.4.2.11 get_matched_subscriptions. Dort wird eine Liste zurueckgegeben; liefert nur den Count, die volle Liste kommt mit Listener-Callbacks.

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

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

Blockiert, bis mindestens min_count Remote-Reader 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.

Verwandt zu OMG DDS 1.4 §2.2.2.4.2.22 wait_for_acknowledgments, aber fokussiert auf Matching statt ACK. Deckt den typischen Producer-Pattern “erst Writer anlegen, dann auf Subscriber warten, dann schreiben” ab.

Errors

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

Examples found in repository

examples/zerodds_perf.rs (line 53)

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

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

pub fn offered_deadline_missed_count(&self) -> u64

Counter fuer offered-Deadline-Verletzungen (Spec §2.2.4.2.9 OFFERED_DEADLINE_MISSED_STATUS). Monoton steigend; steigt um 1 pro abgelaufenem Deadline-Fenster ohne Write. Im Offline-Mode oder bei deadline=INFINITE immer 0.

feuert ggf. on_offered_deadline_missed ueber die Bubble-Up-Kette bei Delta gegenueber dem letzten Aufruf.

pub fn liveliness_lost_count(&self) -> u64

Counter fuer LivelinessLost-Detections (Spec §2.2.4.2.10). Triggert ggf. on_liveliness_lost via Bubble-Up.

pub fn offered_incompatible_qos_status(&self) -> OfferedIncompatibleQosStatus

aktueller OfferedIncompatibleQosStatus (Spec §2.2.4.2.4.2). Triggert ggf. on_offered_incompatible_qos.

pub fn drive_listeners(&self)

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

pub fn assert_liveliness(&self)

Manual-Liveliness-Assert. Spec §2.2.2.4.2.20 assert_liveliness. Setzt den last_liveliness_assert-Timestamp; bei Automatic-Liveliness no-op (jeder write asserts ohnehin).

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

Blockiert, bis alle matched Remote-Reader alle bis jetzt geschriebenen Samples acknowledgt haben, oder timeout abläuft.

Spec: OMG DDS 1.4 §2.2.2.4.2.22 wait_for_acknowledgments. Im Offline-Mode und ohne gematchte Reader sofort Ok(()).

Errors

DdsError::Timeout wenn nicht alle Samples im Zeitfenster acknowledgt sind.

pub fn publication_handle(&self) -> InstanceHandle

Lokaler Publication-Handle dieses DataWriters (Spec §2.2.2.5.1.11). Wird im publication_handle-Feld des SampleInfo mitgegeben. Achtung: das ist NICHT derselbe Handle wie der Entity- InstanceHandle (Spec §2.2.2.1.1) — siehe Self::instance_handle.

pub fn instance_handle(&self) -> InstanceHandle

Spec §2.2.2.1.1 get_instance_handle — Entity-Identifier dieses DataWriters fuer Vergleiche via DomainParticipant::contains_entity.

pub fn instance_tracker(&self) -> InstanceTracker

Liefert den geteilten Instance-Tracker des Writers (Test- und Inspection-Helper, Spec §2.2.2.4.2.5+ Lifecycle-Buchhaltung).

pub fn register_instance(&self, instance: &T) -> Result<InstanceHandle>

Registriert eine Instanz beim DataWriter und liefert ihren stabilen InstanceHandle zurueck. Spec §2.2.2.4.2.5 register_instance.

Fuer non-keyed Topics liefert der Aufruf HANDLE_NIL zurueck (jedes Sample ist seine eigene “Instanz”, die Spec sagt explizit dass register/unregister/dispose hier optional sind).

Errors

Aktuell kann der Aufruf nicht fehlschlagen. Spaeter (Live-Mode) koennen Resource-Limits hier einen OutOfResources-Fehler liefern.

pub fn register_instance_w_timestamp( &self, instance: &T, timestamp: Time, ) -> Result<InstanceHandle>

Wie register_instance, aber mit explizitem Timestamp. Spec §2.2.2.4.2.6.

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.4.2.14 lookup_instance.

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

Entfernt die Instanz aus dem Writer-Set (Spec §2.2.2.4.2.7). Setzt den Lifecycle-Zustand auf NOT_ALIVE_NO_WRITERS, sobald der letzte Writer sich abgemeldet hat.

Errors

BadParameter wenn handle nicht zur Instanz von instance passt (Spec verlangt diese Konsistenz-Pruefung). Wenn handle == HANDLE_NIL, wird der Handle aus instance abgeleitet.

pub fn unregister_instance_w_timestamp( &self, instance: &T, handle: InstanceHandle, timestamp: Time, ) -> Result<()>

Wie unregister_instance, aber mit Timestamp. Spec §2.2.2.4.2.8.

Spec §2.2.3.21 WriterDataLifecycle: wenn autodispose_unregistered_instances=true (Default), wird die Instanz zusaetzlich zum Unregister auch disposed — Reader sehen dann sowohl NOT_ALIVE_DISPOSED als auch NOT_ALIVE_NO_WRITERS.

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

Disposed eine Instanz (Spec §2.2.2.4.2.10). Markiert sie als NOT_ALIVE_DISPOSED; Reader sehen dann ein Sample mit valid_data == false.

Errors

Wie unregister_instance.

pub fn dispose_w_timestamp( &self, instance: &T, handle: InstanceHandle, timestamp: Time, ) -> Result<()>

Wie dispose, aber mit Timestamp. Spec §2.2.2.4.2.11.

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

Gibt den Sample-Wert mit nur den @key-Feldern befuellt zurueck (Spec §2.2.2.4.2.13 get_key_value). Implementierung: wir rekonstruieren T via decode aus dem gespeicherten PLAIN_CDR2-BE-Key-Holder. Damit das funktioniert, muss T::decode einen Key-only-Stream akzeptieren — fuer einfache Records ist das trivialerweise der Fall.

Errors

• BadParameter wenn der Handle unbekannt ist.
• WireError wenn der Key-Holder nicht via T::decode rekonstruierbar ist.

pub fn write_w_timestamp(&self, sample: &T, timestamp: Time) -> Result<()>

Schreibt ein Sample mit explizitem Timestamp (Spec §2.2.2.4.2.16 write_w_timestamp) und aktualisiert die Instanz-Buchhaltung.

Errors

Wie Self::write.

Trait Implementations

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

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

Formats the value using the given formatter.

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

Available on crate feature std only.

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

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

type Qos = DataWriterQos

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.