rustdds 0.3.0

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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146

#[allow(unused_imports)]
use log::{debug,warn,error,trace};

use mio::net::UdpSocket;

use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::io;
use crate::structure::locator::{LocatorKind, LocatorList, Locator};

#[derive(Debug)]
pub struct UDPSender {
  socket: UdpSocket,
}

fn create_socket_to_available_port() -> Option<UdpSocket> {
  for port in 1025..65535 {
    let saddr: SocketAddr = SocketAddr::new("0.0.0.0".parse().unwrap(), port);
    match UdpSocket::bind(&saddr) {
      Ok(l) => return Some(l),
      _ => {}
    }
  }
  None
}

impl UDPSender {
  pub fn new(sender_port: u16) -> UDPSender {
    let saddr: SocketAddr = SocketAddr::new("0.0.0.0".parse().unwrap(), sender_port);
    let socket: UdpSocket = UdpSocket::bind(&saddr).unwrap();

    UDPSender { socket: socket }
  }

  pub fn new_with_random_port() -> UDPSender {
    let socket: UdpSocket = create_socket_to_available_port().unwrap();
    UDPSender { socket: socket }
  }

  pub fn send_to_all(&self, buffer: &[u8], addresses: &Vec<SocketAddr>) {
    for address in addresses.iter() {
      match self.socket.send_to(buffer, address) {
        Ok(_) => (),
        _ => debug!("Unable to send to {}", address),
      };
    }
  }

  pub fn send_to_locator_list(&self, buffer: &[u8], ll: &LocatorList) {
    for loc in ll.iter() {
      self.send_to_locator(buffer,loc)
    }
  }

  pub fn send_to_locator(&self, buffer: &[u8], l: &Locator) {
      match l.kind {
        LocatorKind::LOCATOR_KIND_UDPv4 |
        LocatorKind::LOCATOR_KIND_UDPv6 => {
          let a = SocketAddr::from(l.to_socket_address());
          match self.socket.send_to(buffer, &a) {
            Ok(bytes_sent) =>
              if bytes_sent == buffer.len() { () // ok
              } else {
                error!("send_to_locator - send_to tried {} bytes, sent only {}",
                    buffer.len(), bytes_sent);
              }
            Err(e) => {
              warn!("send_to_locator - send_to {} : {:?}", a, e);
            }
          }
        }
        LocatorKind::LOCATOR_KIND_INVALID |
        LocatorKind::LOCATOR_KIND_RESERVED =>
          error!("send_to_locator: Cannot send to {:?}",l.kind),
        _unknown_kind  =>
          // This is normal, as implementations can define their own kinds.
          trace!("send_to_locator: Unknown LocatorKind: {:?}",l.kind),
      }
  }

  pub fn send_multicast(self, buffer: &[u8], address: Ipv4Addr, port: u16) -> io::Result<usize> {
    if address.is_multicast() {
      let address = SocketAddr::new(IpAddr::V4(address), port);
      return self.socket.send_to(buffer, &SocketAddr::from(address));
    }
    io::Result::Err(io::Error::new(
      io::ErrorKind::Other,
      "Not a multicast address",
    ))
  }

  pub fn send_ipv4_multicast(&self, buffer: &[u8], address: SocketAddr) -> io::Result<usize> {
    if address.ip().is_multicast() {
      return self.socket.send_to(buffer, &address);
    }
    io::Result::Err(io::Error::new(
      io::ErrorKind::Other,
      "Not a multicast address",
    ))
  }
}

#[cfg(test)]
mod tests {
  use super::*;
  use crate::network::udp_listener::*;
  use mio::Token;

  #[test]
  fn udps_single_send() {
    let listener = UDPListener::new(Token(0), "127.0.0.1", 10201);
    let sender = UDPSender::new(11201);

    let data: Vec<u8> = vec![0, 1, 2, 3, 4];

    let addrs = vec![SocketAddr::new("127.0.0.1".parse().unwrap(), 10201)];
    sender.send_to_all(&data, &addrs);

    let rec_data = listener.get_message();

    assert_eq!(rec_data.len(), 5);
    assert_eq!(rec_data, data);
  }

  #[test]
  fn udps_multi_send() {
    let listener_1 = UDPListener::new(Token(0), "127.0.0.1", 10301);
    let listener_2 = UDPListener::new(Token(1), "127.0.0.1", 10302);
    let sender = UDPSender::new(11301);

    let data: Vec<u8> = vec![5, 4, 3, 2, 1, 0];

    let addrs = vec![
      SocketAddr::new("127.0.0.1".parse().unwrap(), 10301),
      SocketAddr::new("127.0.0.1".parse().unwrap(), 10302),
    ];
    sender.send_to_all(&data, &addrs);

    let rec_data_1 = listener_1.get_message();
    let rec_data_2 = listener_2.get_message();

    assert_eq!(rec_data_1.len(), 6);
    assert_eq!(rec_data_1, data);
    assert_eq!(rec_data_2.len(), 6);
    assert_eq!(rec_data_2, data);
  }
}