nodo 0.18.5

A realtime framework for robotics
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

// Copyright 2025 David Weikersdorfer

use crate::{
    channels::{FlushResult, SharedBackStage, TxConnectable, TxSendError},
    core::{AppMonotonicClock, Clock, PubtimeMarker},
    prelude::{Acqtime, DoubleBufferTx, Message, OverflowPolicy, Stamp, Tx},
};

/// Message transmitter using a double-buffer queue
///
/// This is similar to [`DoubleBufferTx<Message<T>>`] with the main difference that
/// [push](MessageTx::push) automatically creates a [Message].
pub struct MessageTx<T> {
    imp: DoubleBufferTx<Message<T>>,
    pub_clock: AppMonotonicClock<PubtimeMarker>,
    seq: u64,
}

impl<T> MessageTx<T> {
    /// Creates a new transmitter which can publish at most the given number of messages per step.
    pub fn new(capacity: usize) -> Self {
        Self::new_imp(DoubleBufferTx::new(capacity))
    }

    /// Creates a new transmitter which can publish an unbounded number of messages per step. Not
    /// recommended as it can lead to queues growing without bound.
    pub fn new_auto_size() -> Self {
        Self::new_imp(DoubleBufferTx::new_auto_size())
    }

    fn new_imp(imp: DoubleBufferTx<Message<T>>) -> Self {
        Self {
            imp,
            pub_clock: AppMonotonicClock::new(),
            seq: 0,
        }
    }

    /// Adds a [Message] to the transmitter with given acqtime and value. The message "pubtime" is
    /// set to the current time and the message sequence number is automatically incremented.
    /// Messages are published after the step functions concludes.
    pub fn push(&mut self, acqtime: Acqtime, value: T) -> Result<(), TxSendError> {
        let seq = self.pub_seq();
        let pubtime = self.pub_clock.now();
        self.imp.push(Message {
            seq,
            stamp: Stamp { acqtime, pubtime },
            value,
        })
    }

    /// Adds multiple messages to the transmitter (see [push](MessageTx::push)).
    pub fn push_many<I: IntoIterator<Item = (Acqtime, T)>>(
        &mut self,
        values: I,
    ) -> Result<(), TxSendError> {
        for (acqtime, value) in values.into_iter() {
            self.push(acqtime, value)?;
        }
        Ok(())
    }

    fn pub_seq(&mut self) -> u64 {
        let seq = self.seq;
        self.seq = self.seq.wrapping_add(1);
        seq
    }
}

impl<T: Send + Sync + Clone> Tx for MessageTx<T> {
    fn flush(&mut self) -> FlushResult {
        self.imp.flush()
    }

    fn is_connected(&self) -> bool {
        self.imp.is_connected()
    }

    fn len(&self) -> usize {
        self.imp.len()
    }
}

impl<V: Send + Sync + Clone> TxConnectable for MessageTx<V> {
    type Message = Message<V>;

    fn has_max_connection_count(&self) -> bool {
        self.imp.has_max_connection_count()
    }

    fn overflow_policy(&self) -> OverflowPolicy {
        self.imp.overflow_policy()
    }

    fn on_connect(&mut self, stage: SharedBackStage<Self::Message>) {
        self.imp.on_connect(stage);
    }
}