oxictl 0.1.1

Pure Rust Real-Time Control Systems Framework
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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135

//! ReaderProxy — per-reader ACK state tracked by a StatefulWriter.

use crate::protocol::dds::message::submessage::AckNack;
use crate::protocol::dds::types::guid::Guid;
use crate::protocol::dds::types::locator::Locator;
use crate::protocol::dds::types::sequence::SequenceNumber;

/// Tracks per-reader ACK state inside a [`StatefulWriter`].
///
/// The proxy records the highest acknowledged SN and any SNs the reader
/// has explicitly requested retransmission of via NACK bitmask.
///
/// [`StatefulWriter`]: super::writer::StatefulWriter
pub struct ReaderProxy {
    /// Full GUID of the remote reader.
    pub remote_reader_guid: Guid,
    /// Unicast locators used to deliver DATA and HEARTBEAT.
    pub unicast_locators: Vec<Locator>,
    /// Highest SN the reader has fully acknowledged (everything below
    /// `reader_sn_state.bitmap_base` in the most recent ACKNACK).
    pub acked_sn: Option<SequenceNumber>,
    /// SNs the reader explicitly requested retransmit via NACK bitmap.
    pub requested_changes: Vec<SequenceNumber>,
}

impl ReaderProxy {
    /// Create a new proxy with no acknowledgements.
    pub fn new(guid: Guid, unicast_locators: Vec<Locator>) -> Self {
        Self {
            remote_reader_guid: guid,
            unicast_locators,
            acked_sn: None,
            requested_changes: Vec::new(),
        }
    }

    /// Process an incoming ACKNACK submessage from this reader.
    ///
    /// Updates `acked_sn` to `bitmap_base - 1` (everything below the base is
    /// acknowledged) and populates `requested_changes` with the SNs whose bits
    /// are SET in the bitmap (those are the gaps the reader wants retransmitted).
    ///
    /// Returns `true` if there are outstanding changes to retransmit.
    pub fn process_acknack(&mut self, acknack: &AckNack) -> bool {
        // The reader has acknowledged everything strictly before bitmap_base.
        let base_v = acknack.reader_sn_state.bitmap_base.to_i64();
        if base_v > 1 {
            let acked = SequenceNumber::new(base_v - 1);
            self.acked_sn = Some(match self.acked_sn {
                None => acked,
                Some(prev) => {
                    if acked > prev {
                        acked
                    } else {
                        prev
                    }
                }
            });
        }

        // Collect all SNs that are SET in the bitmap — those are missing/requested.
        self.requested_changes.clear();
        for sn in acknack.reader_sn_state.iter() {
            if !self.requested_changes.contains(&sn) {
                self.requested_changes.push(sn);
            }
        }

        !self.requested_changes.is_empty()
    }

    /// Drain (remove and return) all currently requested changes.
    ///
    /// Called by the writer before transmitting retransmissions.
    pub fn drain_requested(&mut self) -> Vec<SequenceNumber> {
        let mut out = Vec::new();
        core::mem::swap(&mut self.requested_changes, &mut out);
        out
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::protocol::dds::byte_cursor::Endianness;
    use crate::protocol::dds::message::submessage::AckNack;
    use crate::protocol::dds::types::guid::{EntityId, GuidPrefix, ENTITYID_UNKNOWN};
    use crate::protocol::dds::types::sequence::{SequenceNumber, SequenceNumberSet};

    fn sn(v: i64) -> SequenceNumber {
        SequenceNumber::new(v)
    }

    fn make_proxy() -> ReaderProxy {
        let guid = Guid {
            prefix: GuidPrefix([0xBBu8; 12]),
            entity_id: EntityId {
                entity_key: [0, 0, 7],
                entity_kind: 0x04,
            },
        };
        ReaderProxy::new(guid, Vec::new())
    }

    #[test]
    fn reader_proxy_process_acknack() {
        let mut proxy = make_proxy();

        // Build an AckNack: bitmap_base=3, bits for SN(3) and SN(5) are set.
        // SN(3) is at bit-offset 0 (base=3), SN(5) is at bit-offset 2.
        let mut sn_set = SequenceNumberSet::empty(sn(3));
        sn_set.set(sn(3)).unwrap();
        sn_set.set(sn(5)).unwrap();

        let acknack = AckNack {
            endianness: Endianness::Little,
            final_flag: true,
            reader_id: ENTITYID_UNKNOWN,
            writer_id: ENTITYID_UNKNOWN,
            reader_sn_state: sn_set,
            count: 1,
        };

        let has_changes = proxy.process_acknack(&acknack);
        assert!(has_changes);

        // acked_sn = bitmap_base - 1 = 3 - 1 = 2
        assert_eq!(proxy.acked_sn, Some(sn(2)));

        // requested_changes must contain SN(3) and SN(5).
        let mut req = proxy.requested_changes.clone();
        req.sort();
        assert_eq!(req, vec![sn(3), sn(5)]);
    }
}