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