src/libstd/net/udp.rs

   1 // Copyright 2015 The Rust Project Developers. See the COPYRIGHT
   2 // file at the top-level directory of this distribution and at
   3 // http://rust-lang.org/COPYRIGHT.
   4 //
   5 // Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
   6 // http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
   7 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
   8 // option. This file may not be copied, modified, or distributed
   9 // except according to those terms.
  10
  11 use fmt;
  12 use io::{self, Error, ErrorKind};
  13 use net::{ToSocketAddrs, SocketAddr};
  14 use sys_common::net as net_imp;
  15 use sys_common::{AsInner, FromInner, IntoInner};
  16 use time::Duration;
  17
  18 /// A User Datagram Protocol socket.
  19 ///
  20 /// This is an implementation of a bound UDP socket. This supports both IPv4 and
  21 /// IPv6 addresses, and there is no corresponding notion of a server because UDP
  22 /// is a datagram protocol.
  23 ///
  24 /// # Examples
  25 ///
  26 /// ```no_run
  27 /// use std::net::UdpSocket;
  28 ///
  29 /// # fn foo() -> std::io::Result<()> {
  30 /// let mut socket = try!(UdpSocket::bind("127.0.0.1:34254"));
  31 ///
  32 /// let mut buf = [0; 10];
  33 /// let (amt, src) = try!(socket.recv_from(&mut buf));
  34 ///
  35 /// // Send a reply to the socket we received data from
  36 /// let buf = &mut buf[..amt];
  37 /// buf.reverse();
  38 /// try!(socket.send_to(buf, &src));
  39 ///
  40 /// drop(socket); // close the socket
  41 /// # Ok(())
  42 /// # }
  43 /// ```
  44 #[stable(feature = "rust1", since = "1.0.0")]
  45 pub struct UdpSocket(net_imp::UdpSocket);
  46
  47 impl UdpSocket {
  48     /// Creates a UDP socket from the given address.
  49     ///
  50     /// The address type can be any implementor of `ToSocketAddr` trait. See
  51     /// its documentation for concrete examples.
  52     #[stable(feature = "rust1", since = "1.0.0")]
  53     pub fn bind<A: ToSocketAddrs>(addr: A) -> io::Result<UdpSocket> {
  54         super::each_addr(addr, net_imp::UdpSocket::bind).map(UdpSocket)
  55     }
  56
  57     /// Receives data from the socket. On success, returns the number of bytes
  58     /// read and the address from whence the data came.
  59     #[stable(feature = "rust1", since = "1.0.0")]
  60     pub fn recv_from(&self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
  61         self.0.recv_from(buf)
  62     }
  63
  64     /// Sends data on the socket to the given address. On success, returns the
  65     /// number of bytes written.
  66     ///
  67     /// Address type can be any implementor of `ToSocketAddrs` trait. See its
  68     /// documentation for concrete examples.
  69     #[stable(feature = "rust1", since = "1.0.0")]
  70     pub fn send_to<A: ToSocketAddrs>(&self, buf: &[u8], addr: A)
  71                                      -> io::Result<usize> {
  72         match try!(addr.to_socket_addrs()).next() {
  73             Some(addr) => self.0.send_to(buf, &addr),
  74             None => Err(Error::new(ErrorKind::InvalidInput,
  75                                    "no addresses to send data to")),
  76         }
  77     }
  78
  79     /// Returns the socket address that this socket was created from.
  80     #[stable(feature = "rust1", since = "1.0.0")]
  81     pub fn local_addr(&self) -> io::Result<SocketAddr> {
  82         self.0.socket_addr()
  83     }
  84
  85     /// Creates a new independently owned handle to the underlying socket.
  86     ///
  87     /// The returned `UdpSocket` is a reference to the same socket that this
  88     /// object references. Both handles will read and write the same port, and
  89     /// options set on one socket will be propagated to the other.
  90     #[stable(feature = "rust1", since = "1.0.0")]
  91     pub fn try_clone(&self) -> io::Result<UdpSocket> {
  92         self.0.duplicate().map(UdpSocket)
  93     }
  94
  95     /// Sets the read timeout to the timeout specified.
  96     ///
  97     /// If the value specified is `None`, then `read` calls will block
  98     /// indefinitely. It is an error to pass the zero `Duration` to this
  99     /// method.
 100     ///
 101     /// # Note
 102     ///
 103     /// Platforms may return a different error code whenever a read times out as
 104     /// a result of setting this option. For example Unix typically returns an
 105     /// error of the kind `WouldBlock`, but Windows may return `TimedOut`.
 106     #[stable(feature = "socket_timeout", since = "1.4.0")]
 107     pub fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
 108         self.0.set_read_timeout(dur)
 109     }
 110
 111     /// Sets the write timeout to the timeout specified.
 112     ///
 113     /// If the value specified is `None`, then `write` calls will block
 114     /// indefinitely. It is an error to pass the zero `Duration` to this
 115     /// method.
 116     ///
 117     /// # Note
 118     ///
 119     /// Platforms may return a different error code whenever a write times out
 120     /// as a result of setting this option. For example Unix typically returns
 121     /// an error of the kind `WouldBlock`, but Windows may return `TimedOut`.
 122     #[stable(feature = "socket_timeout", since = "1.4.0")]
 123     pub fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
 124         self.0.set_write_timeout(dur)
 125     }
 126
 127     /// Returns the read timeout of this socket.
 128     ///
 129     /// If the timeout is `None`, then `read` calls will block indefinitely.
 130     #[stable(feature = "socket_timeout", since = "1.4.0")]
 131     pub fn read_timeout(&self) -> io::Result<Option<Duration>> {
 132         self.0.read_timeout()
 133     }
 134
 135     /// Returns the write timeout of this socket.
 136     ///
 137     /// If the timeout is `None`, then `write` calls will block indefinitely.
 138     #[stable(feature = "socket_timeout", since = "1.4.0")]
 139     pub fn write_timeout(&self) -> io::Result<Option<Duration>> {
 140         self.0.write_timeout()
 141     }
 142 }
 143
 144 impl AsInner<net_imp::UdpSocket> for UdpSocket {
 145     fn as_inner(&self) -> &net_imp::UdpSocket { &self.0 }
 146 }
 147
 148 impl FromInner<net_imp::UdpSocket> for UdpSocket {
 149     fn from_inner(inner: net_imp::UdpSocket) -> UdpSocket { UdpSocket(inner) }
 150 }
 151
 152 impl IntoInner<net_imp::UdpSocket> for UdpSocket {
 153     fn into_inner(self) -> net_imp::UdpSocket { self.0 }
 154 }
 155
 156 impl fmt::Debug for UdpSocket {
 157     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
 158         self.0.fmt(f)
 159     }
 160 }
 161
 162 #[cfg(test)]
 163 mod tests {
 164     use prelude::v1::*;
 165
 166     use io::ErrorKind;
 167     use net::*;
 168     use net::test::{next_test_ip4, next_test_ip6};
 169     use sync::mpsc::channel;
 170     use sys_common::AsInner;
 171     use time::Duration;
 172     use thread;
 173
 174     fn each_ip(f: &mut FnMut(SocketAddr, SocketAddr)) {
 175         f(next_test_ip4(), next_test_ip4());
 176         f(next_test_ip6(), next_test_ip6());
 177     }
 178
 179     macro_rules! t {
 180         ($e:expr) => {
 181             match $e {
 182                 Ok(t) => t,
 183                 Err(e) => panic!("received error for `{}`: {}", stringify!($e), e),
 184             }
 185         }
 186     }
 187
 188     #[test]
 189     fn bind_error() {
 190         match UdpSocket::bind("1.1.1.1:9999") {
 191             Ok(..) => panic!(),
 192             Err(e) => {
 193                 assert_eq!(e.kind(), ErrorKind::AddrNotAvailable)
 194             }
 195         }
 196     }
 197
 198     #[test]
 199     fn socket_smoke_test_ip4() {
 200         each_ip(&mut |server_ip, client_ip| {
 201             let (tx1, rx1) = channel();
 202             let (tx2, rx2) = channel();
 203
 204             let _t = thread::spawn(move|| {
 205                 let client = t!(UdpSocket::bind(&client_ip));
 206                 rx1.recv().unwrap();
 207                 t!(client.send_to(&[99], &server_ip));
 208                 tx2.send(()).unwrap();
 209             });
 210
 211             let server = t!(UdpSocket::bind(&server_ip));
 212             tx1.send(()).unwrap();
 213             let mut buf = [0];
 214             let (nread, src) = t!(server.recv_from(&mut buf));
 215             assert_eq!(nread, 1);
 216             assert_eq!(buf[0], 99);
 217             assert_eq!(src, client_ip);
 218             rx2.recv().unwrap();
 219         })
 220     }
 221
 222     #[test]
 223     fn socket_name_ip4() {
 224         each_ip(&mut |addr, _| {
 225             let server = t!(UdpSocket::bind(&addr));
 226             assert_eq!(addr, t!(server.local_addr()));
 227         })
 228     }
 229
 230     #[test]
 231     fn udp_clone_smoke() {
 232         each_ip(&mut |addr1, addr2| {
 233             let sock1 = t!(UdpSocket::bind(&addr1));
 234             let sock2 = t!(UdpSocket::bind(&addr2));
 235
 236             let _t = thread::spawn(move|| {
 237                 let mut buf = [0, 0];
 238                 assert_eq!(sock2.recv_from(&mut buf).unwrap(), (1, addr1));
 239                 assert_eq!(buf[0], 1);
 240                 t!(sock2.send_to(&[2], &addr1));
 241             });
 242
 243             let sock3 = t!(sock1.try_clone());
 244
 245             let (tx1, rx1) = channel();
 246             let (tx2, rx2) = channel();
 247             let _t = thread::spawn(move|| {
 248                 rx1.recv().unwrap();
 249                 t!(sock3.send_to(&[1], &addr2));
 250                 tx2.send(()).unwrap();
 251             });
 252             tx1.send(()).unwrap();
 253             let mut buf = [0, 0];
 254             assert_eq!(sock1.recv_from(&mut buf).unwrap(), (1, addr2));
 255             rx2.recv().unwrap();
 256         })
 257     }
 258
 259     #[test]
 260     fn udp_clone_two_read() {
 261         each_ip(&mut |addr1, addr2| {
 262             let sock1 = t!(UdpSocket::bind(&addr1));
 263             let sock2 = t!(UdpSocket::bind(&addr2));
 264             let (tx1, rx) = channel();
 265             let tx2 = tx1.clone();
 266
 267             let _t = thread::spawn(move|| {
 268                 t!(sock2.send_to(&[1], &addr1));
 269                 rx.recv().unwrap();
 270                 t!(sock2.send_to(&[2], &addr1));
 271                 rx.recv().unwrap();
 272             });
 273
 274             let sock3 = t!(sock1.try_clone());
 275
 276             let (done, rx) = channel();
 277             let _t = thread::spawn(move|| {
 278                 let mut buf = [0, 0];
 279                 t!(sock3.recv_from(&mut buf));
 280                 tx2.send(()).unwrap();
 281                 done.send(()).unwrap();
 282             });
 283             let mut buf = [0, 0];
 284             t!(sock1.recv_from(&mut buf));
 285             tx1.send(()).unwrap();
 286
 287             rx.recv().unwrap();
 288         })
 289     }
 290
 291     #[test]
 292     fn udp_clone_two_write() {
 293         each_ip(&mut |addr1, addr2| {
 294             let sock1 = t!(UdpSocket::bind(&addr1));
 295             let sock2 = t!(UdpSocket::bind(&addr2));
 296
 297             let (tx, rx) = channel();
 298             let (serv_tx, serv_rx) = channel();
 299
 300             let _t = thread::spawn(move|| {
 301                 let mut buf = [0, 1];
 302                 rx.recv().unwrap();
 303                 t!(sock2.recv_from(&mut buf));
 304                 serv_tx.send(()).unwrap();
 305             });
 306
 307             let sock3 = t!(sock1.try_clone());
 308
 309             let (done, rx) = channel();
 310             let tx2 = tx.clone();
 311             let _t = thread::spawn(move|| {
 312                 match sock3.send_to(&[1], &addr2) {
 313                     Ok(..) => { let _ = tx2.send(()); }
 314                     Err(..) => {}
 315                 }
 316                 done.send(()).unwrap();
 317             });
 318             match sock1.send_to(&[2], &addr2) {
 319                 Ok(..) => { let _ = tx.send(()); }
 320                 Err(..) => {}
 321             }
 322             drop(tx);
 323
 324             rx.recv().unwrap();
 325             serv_rx.recv().unwrap();
 326         })
 327     }
 328
 329     #[test]
 330     fn debug() {
 331         let name = if cfg!(windows) {"socket"} else {"fd"};
 332         let socket_addr = next_test_ip4();
 333
 334         let udpsock = t!(UdpSocket::bind(&socket_addr));
 335         let udpsock_inner = udpsock.0.socket().as_inner();
 336         let compare = format!("UdpSocket {{ addr: {:?}, {}: {:?} }}",
 337                               socket_addr, name, udpsock_inner);
 338         assert_eq!(format!("{:?}", udpsock), compare);
 339     }
 340
 341     // FIXME: re-enabled bitrig/openbsd/netbsd tests once their socket timeout code
 342     //        no longer has rounding errors.
 343     #[cfg_attr(any(target_os = "bitrig", target_os = "netbsd", target_os = "openbsd"), ignore)]
 344     #[test]
 345     fn timeouts() {
 346         let addr = next_test_ip4();
 347
 348         let stream = t!(UdpSocket::bind(&addr));
 349         let dur = Duration::new(15410, 0);
 350
 351         assert_eq!(None, t!(stream.read_timeout()));
 352
 353         t!(stream.set_read_timeout(Some(dur)));
 354         assert_eq!(Some(dur), t!(stream.read_timeout()));
 355
 356         assert_eq!(None, t!(stream.write_timeout()));
 357
 358         t!(stream.set_write_timeout(Some(dur)));
 359         assert_eq!(Some(dur), t!(stream.write_timeout()));
 360
 361         t!(stream.set_read_timeout(None));
 362         assert_eq!(None, t!(stream.read_timeout()));
 363
 364         t!(stream.set_write_timeout(None));
 365         assert_eq!(None, t!(stream.write_timeout()));
 366     }
 367
 368     #[test]
 369     fn test_read_timeout() {
 370         let addr = next_test_ip4();
 371
 372         let mut stream = t!(UdpSocket::bind(&addr));
 373         t!(stream.set_read_timeout(Some(Duration::from_millis(1000))));
 374
 375         let mut buf = [0; 10];
 376         let wait = Duration::span(|| {
 377             let kind = stream.recv_from(&mut buf).err().expect("expected error").kind();
 378             assert!(kind == ErrorKind::WouldBlock || kind == ErrorKind::TimedOut);
 379         });
 380         assert!(wait > Duration::from_millis(400));
 381     }
 382
 383     #[test]
 384     fn test_read_with_timeout() {
 385         let addr = next_test_ip4();
 386
 387         let mut stream = t!(UdpSocket::bind(&addr));
 388         t!(stream.set_read_timeout(Some(Duration::from_millis(1000))));
 389
 390         t!(stream.send_to(b"hello world", &addr));
 391
 392         let mut buf = [0; 11];
 393         t!(stream.recv_from(&mut buf));
 394         assert_eq!(b"hello world", &buf[..]);
 395
 396         let wait = Duration::span(|| {
 397             let kind = stream.recv_from(&mut buf).err().expect("expected error").kind();
 398             assert!(kind == ErrorKind::WouldBlock || kind == ErrorKind::TimedOut);
 399         });
 400         assert!(wait > Duration::from_millis(400));
 401     }
 402 }