dust_dds 0.15.0

Data Distribution Service (DDS) implementation
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

use crate::{
    rtps::history_cache::HistoryCache,
    rtps_messages::{self, submessages::data::DataSubmessage},
    transport::types::{CacheChange, ENTITYID_UNKNOWN, Guid, GuidPrefix},
};
use alloc::boxed::Box;

pub struct RtpsStatelessReader {
    guid: Guid,
    history_cache: Box<dyn HistoryCache + Send + Sync>,
}

impl RtpsStatelessReader {
    pub fn new(guid: Guid, history_cache: Box<dyn HistoryCache + Send + Sync>) -> Self {
        Self {
            guid,
            history_cache,
        }
    }

    pub fn guid(&self) -> Guid {
        self.guid
    }

    pub async fn on_data_submessage_received(
        &mut self,
        data_submessage: &DataSubmessage,
        source_guid_prefix: GuidPrefix,
        source_timestamp: Option<rtps_messages::types::Time>,
    ) {
        if data_submessage.reader_id() == ENTITYID_UNKNOWN
            || data_submessage.reader_id() == self.guid.entity_id()
        {
            if let Ok(change) = CacheChange::try_from_data_submessage(
                data_submessage,
                source_guid_prefix,
                source_timestamp,
            ) {
                // Stateless reader behavior. We add the change if the data is correct. No error is printed
                // because all readers would get changes marked with ENTITYID_UNKNOWN
                self.history_cache.add_change(change).await;
            }
        }
    }
}