rustdds 0.11.8

Native Rust DDS implementation with RTPS
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

use std::{cmp::min, io};

use bytes::{Bytes, BytesMut};
use speedy::{Context, Writable, Writer};
use byteorder::ReadBytesExt;
use log::warn;

use crate::RepresentationIdentifier;

/// A SerializedPayload submessage element contains the serialized
/// representation of either value of an application-defined data-object or
/// the value of the key that uniquely identifies the data-object
/// See RTPS spec v2.3 section 10.
/// Standard representation identifier values are defined in sections 10.2 -
/// 10.5 representation_options "shall be interpreted in the context of the
/// RepresentationIdentifier, such that each RepresentationIdentifier may define
/// the representation_options that it requires." and "The [2.3] version of the
/// protocol does not use the representation_options: The sender shall set the
/// representation_options to zero. The receiver shall ignore the value of the
/// representation_options."
#[derive(Debug, PartialEq, Eq, Clone)]
pub struct SerializedPayload {
  pub representation_identifier: RepresentationIdentifier,
  // Can represent payload protection kind. Currently not used outside security.
  pub representation_options: [u8; 2],
  pub value: Bytes,
}

// header length
// 2 bytes for representation identifier
// + 2 bytes for representation options
const H_LEN: usize = 2 + 2;

impl SerializedPayload {
  pub fn new(rep_id: RepresentationIdentifier, payload: Vec<u8>) -> Self {
    Self {
      representation_identifier: rep_id,
      representation_options: [0, 0],
      value: Bytes::from(payload),
    }
  }

  pub fn new_from_bytes(rep_id: RepresentationIdentifier, payload: Bytes) -> Self {
    Self {
      representation_identifier: rep_id,
      representation_options: [0, 0],
      value: payload,
    }
  }

  pub(crate) fn reallocate(&mut self) {
    self.value = Bytes::copy_from_slice(&self.value);
  }

  /// serialized size in bytes
  pub fn len_serialized(&self) -> usize {
    H_LEN + self.value.len()
  }

  // a slice of serialized data
  // This has a lot of H_LEN offsets, because the data to be sliced
  // is header + value
  pub fn bytes_slice(&self, from: usize, to_before: usize) -> Bytes {
    // sanitize inputs. These are unsigned values, so always at least zero.
    let to_before = min(to_before, self.value.len() + H_LEN);
    let from = min(from, to_before);

    if from >= H_LEN {
      // no need to copy, can return a slice
      self.value.slice(from - H_LEN..to_before - H_LEN)
    } else {
      // We need to copy the payload on order to prefix with header
      let mut b = BytesMut::with_capacity(to_before);
      b.extend_from_slice(&self.representation_identifier.bytes);
      b.extend_from_slice(&self.representation_options);
      assert_eq!(b.len(), H_LEN);
      if to_before > H_LEN {
        b.extend_from_slice(&self.value.slice(..to_before - H_LEN));
      }
      b.freeze().slice(from..to_before)
    }
  }

  // Implement deserialization here, because Speedy just makes it difficult.
  pub fn from_bytes(bytes: &Bytes) -> io::Result<Self> {
    let mut reader = io::Cursor::new(&bytes);
    let representation_identifier = RepresentationIdentifier {
      bytes: [reader.read_u8()?, reader.read_u8()?],
    };
    let representation_options = [reader.read_u8()?, reader.read_u8()?];
    let value = if bytes.len() >= H_LEN {
      bytes.slice(H_LEN..)
    } else {
      warn!("DATA submessage was smaller than submessage header: {bytes:?}");
      return Err(io::Error::new(
        io::ErrorKind::Other,
        "Too short DATA submessage.",
      ));
    };

    Ok(Self {
      representation_identifier,
      representation_options,
      value,
    })
  }
}

impl<C: Context> Writable<C> for SerializedPayload {
  fn write_to<T: ?Sized + Writer<C>>(&self, writer: &mut T) -> Result<(), C::Error> {
    writer.write_u8(self.representation_identifier.bytes[0])?;
    writer.write_u8(self.representation_identifier.bytes[1])?;
    writer.write_u8(self.representation_options[0])?;
    writer.write_u8(self.representation_options[1])?;
    writer.write_bytes(&self.value)?;
    Ok(())
  }
}

// TODO: Should this be fallible try_from instead?
// The speedy write_to_buffer() call could in theory fail, but it is hard to see
// how.
impl From<SerializedPayload> for Bytes {
  fn from(sp: SerializedPayload) -> Bytes {
    let buf = sp.write_to_vec().unwrap(); //TODO: deal with (unlikely) error
    buf.into()
  }
}