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 #[stable(feature = "rust1", since = "1.0.0")]
 157 impl fmt::Debug for UdpSocket {
 158     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
 159         self.0.fmt(f)
 160     }
 161 }
 162
 163 #[cfg(test)]
 164 mod tests {
 165     use prelude::v1::*;
 166
 167     use io::ErrorKind;
 168     use net::*;
 169     use net::test::{next_test_ip4, next_test_ip6};
 170     use sync::mpsc::channel;
 171     use sys_common::AsInner;
 172     use time::Duration;
 173     use thread;
 174
 175     fn each_ip(f: &mut FnMut(SocketAddr, SocketAddr)) {
 176         f(next_test_ip4(), next_test_ip4());
 177         f(next_test_ip6(), next_test_ip6());
 178     }
 179
 180     macro_rules! t {
 181         ($e:expr) => {
 182             match $e {
 183                 Ok(t) => t,
 184                 Err(e) => panic!("received error for `{}`: {}", stringify!($e), e),
 185             }
 186         }
 187     }
 188
 189     #[test]
 190     fn bind_error() {
 191         match UdpSocket::bind("1.1.1.1:9999") {
 192             Ok(..) => panic!(),
 193             Err(e) => {
 194                 assert_eq!(e.kind(), ErrorKind::AddrNotAvailable)
 195             }
 196         }
 197     }
 198
 199     #[test]
 200     fn socket_smoke_test_ip4() {
 201         each_ip(&mut |server_ip, client_ip| {
 202             let (tx1, rx1) = channel();
 203             let (tx2, rx2) = channel();
 204
 205             let _t = thread::spawn(move|| {
 206                 let client = t!(UdpSocket::bind(&client_ip));
 207                 rx1.recv().unwrap();
 208                 t!(client.send_to(&[99], &server_ip));
 209                 tx2.send(()).unwrap();
 210             });
 211
 212             let server = t!(UdpSocket::bind(&server_ip));
 213             tx1.send(()).unwrap();
 214             let mut buf = [0];
 215             let (nread, src) = t!(server.recv_from(&mut buf));
 216             assert_eq!(nread, 1);
 217             assert_eq!(buf[0], 99);
 218             assert_eq!(src, client_ip);
 219             rx2.recv().unwrap();
 220         })
 221     }
 222
 223     #[test]
 224     fn socket_name_ip4() {
 225         each_ip(&mut |addr, _| {
 226             let server = t!(UdpSocket::bind(&addr));
 227             assert_eq!(addr, t!(server.local_addr()));
 228         })
 229     }
 230
 231     #[test]
 232     fn udp_clone_smoke() {
 233         each_ip(&mut |addr1, addr2| {
 234             let sock1 = t!(UdpSocket::bind(&addr1));
 235             let sock2 = t!(UdpSocket::bind(&addr2));
 236
 237             let _t = thread::spawn(move|| {
 238                 let mut buf = [0, 0];
 239                 assert_eq!(sock2.recv_from(&mut buf).unwrap(), (1, addr1));
 240                 assert_eq!(buf[0], 1);
 241                 t!(sock2.send_to(&[2], &addr1));
 242             });
 243
 244             let sock3 = t!(sock1.try_clone());
 245
 246             let (tx1, rx1) = channel();
 247             let (tx2, rx2) = channel();
 248             let _t = thread::spawn(move|| {
 249                 rx1.recv().unwrap();
 250                 t!(sock3.send_to(&[1], &addr2));
 251                 tx2.send(()).unwrap();
 252             });
 253             tx1.send(()).unwrap();
 254             let mut buf = [0, 0];
 255             assert_eq!(sock1.recv_from(&mut buf).unwrap(), (1, addr2));
 256             rx2.recv().unwrap();
 257         })
 258     }
 259
 260     #[test]
 261     fn udp_clone_two_read() {
 262         each_ip(&mut |addr1, addr2| {
 263             let sock1 = t!(UdpSocket::bind(&addr1));
 264             let sock2 = t!(UdpSocket::bind(&addr2));
 265             let (tx1, rx) = channel();
 266             let tx2 = tx1.clone();
 267
 268             let _t = thread::spawn(move|| {
 269                 t!(sock2.send_to(&[1], &addr1));
 270                 rx.recv().unwrap();
 271                 t!(sock2.send_to(&[2], &addr1));
 272                 rx.recv().unwrap();
 273             });
 274
 275             let sock3 = t!(sock1.try_clone());
 276
 277             let (done, rx) = channel();
 278             let _t = thread::spawn(move|| {
 279                 let mut buf = [0, 0];
 280                 t!(sock3.recv_from(&mut buf));
 281                 tx2.send(()).unwrap();
 282                 done.send(()).unwrap();
 283             });
 284             let mut buf = [0, 0];
 285             t!(sock1.recv_from(&mut buf));
 286             tx1.send(()).unwrap();
 287
 288             rx.recv().unwrap();
 289         })
 290     }
 291
 292     #[test]
 293     fn udp_clone_two_write() {
 294         each_ip(&mut |addr1, addr2| {
 295             let sock1 = t!(UdpSocket::bind(&addr1));
 296             let sock2 = t!(UdpSocket::bind(&addr2));
 297
 298             let (tx, rx) = channel();
 299             let (serv_tx, serv_rx) = channel();
 300
 301             let _t = thread::spawn(move|| {
 302                 let mut buf = [0, 1];
 303                 rx.recv().unwrap();
 304                 t!(sock2.recv_from(&mut buf));
 305                 serv_tx.send(()).unwrap();
 306             });
 307
 308             let sock3 = t!(sock1.try_clone());
 309
 310             let (done, rx) = channel();
 311             let tx2 = tx.clone();
 312             let _t = thread::spawn(move|| {
 313                 match sock3.send_to(&[1], &addr2) {
 314                     Ok(..) => { let _ = tx2.send(()); }
 315                     Err(..) => {}
 316                 }
 317                 done.send(()).unwrap();
 318             });
 319             match sock1.send_to(&[2], &addr2) {
 320                 Ok(..) => { let _ = tx.send(()); }
 321                 Err(..) => {}
 322             }
 323             drop(tx);
 324
 325             rx.recv().unwrap();
 326             serv_rx.recv().unwrap();
 327         })
 328     }
 329
 330     #[test]
 331     fn debug() {
 332         let name = if cfg!(windows) {"socket"} else {"fd"};
 333         let socket_addr = next_test_ip4();
 334
 335         let udpsock = t!(UdpSocket::bind(&socket_addr));
 336         let udpsock_inner = udpsock.0.socket().as_inner();
 337         let compare = format!("UdpSocket {{ addr: {:?}, {}: {:?} }}",
 338                               socket_addr, name, udpsock_inner);
 339         assert_eq!(format!("{:?}", udpsock), compare);
 340     }
 341
 342     // FIXME: re-enabled bitrig/openbsd/netbsd tests once their socket timeout code
 343     //        no longer has rounding errors.
 344     #[cfg_attr(any(target_os = "bitrig", target_os = "netbsd", target_os = "openbsd"), ignore)]
 345     #[test]
 346     fn timeouts() {
 347         let addr = next_test_ip4();
 348
 349         let stream = t!(UdpSocket::bind(&addr));
 350         let dur = Duration::new(15410, 0);
 351
 352         assert_eq!(None, t!(stream.read_timeout()));
 353
 354         t!(stream.set_read_timeout(Some(dur)));
 355         assert_eq!(Some(dur), t!(stream.read_timeout()));
 356
 357         assert_eq!(None, t!(stream.write_timeout()));
 358
 359         t!(stream.set_write_timeout(Some(dur)));
 360         assert_eq!(Some(dur), t!(stream.write_timeout()));
 361
 362         t!(stream.set_read_timeout(None));
 363         assert_eq!(None, t!(stream.read_timeout()));
 364
 365         t!(stream.set_write_timeout(None));
 366         assert_eq!(None, t!(stream.write_timeout()));
 367     }
 368
 369     #[test]
 370     fn test_read_timeout() {
 371         let addr = next_test_ip4();
 372
 373         let mut stream = t!(UdpSocket::bind(&addr));
 374         t!(stream.set_read_timeout(Some(Duration::from_millis(1000))));
 375
 376         let mut buf = [0; 10];
 377         let wait = Duration::span(|| {
 378             let kind = stream.recv_from(&mut buf).err().expect("expected error").kind();
 379             assert!(kind == ErrorKind::WouldBlock || kind == ErrorKind::TimedOut);
 380         });
 381         assert!(wait > Duration::from_millis(400));
 382     }
 383
 384     #[test]
 385     fn test_read_with_timeout() {
 386         let addr = next_test_ip4();
 387
 388         let mut stream = t!(UdpSocket::bind(&addr));
 389         t!(stream.set_read_timeout(Some(Duration::from_millis(1000))));
 390
 391         t!(stream.send_to(b"hello world", &addr));
 392
 393         let mut buf = [0; 11];
 394         t!(stream.recv_from(&mut buf));
 395         assert_eq!(b"hello world", &buf[..]);
 396
 397         let wait = Duration::span(|| {
 398             let kind = stream.recv_from(&mut buf).err().expect("expected error").kind();
 399             assert!(kind == ErrorKind::WouldBlock || kind == ErrorKind::TimedOut);
 400         });
 401         assert!(wait > Duration::from_millis(400));
 402     }
 403 }