src/libstd/io/net/udp.rs

   1 // Copyright 2013 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 //! UDP (User Datagram Protocol) network connections.
  12 //!
  13 //! This module contains the ability to open a UDP stream to a socket address.
  14 //! The destination and binding addresses can either be an IPv4 or IPv6
  15 //! address. There is no corresponding notion of a server because UDP is a
  16 //! datagram protocol.
  17
  18 use clone::Clone;
  19 use io::net::ip::{SocketAddr, IpAddr, ToSocketAddr};
  20 use io::{Reader, Writer, IoResult};
  21 use ops::FnOnce;
  22 use option::Option;
  23 use result::Result::{Ok, Err};
  24 use sys::udp::UdpSocket as UdpSocketImp;
  25 use sys_common;
  26
  27 /// A User Datagram Protocol socket.
  28 ///
  29 /// This is an implementation of a bound UDP socket. This supports both IPv4 and
  30 /// IPv6 addresses, and there is no corresponding notion of a server because UDP
  31 /// is a datagram protocol.
  32 ///
  33 /// # Example
  34 ///
  35 /// ```rust,no_run
  36 /// # #![allow(unused_must_use)]
  37 /// #![feature(slicing_syntax)]
  38 ///
  39 /// use std::io::net::udp::UdpSocket;
  40 /// use std::io::net::ip::{Ipv4Addr, SocketAddr};
  41 /// fn main() {
  42 ///     let addr = SocketAddr { ip: Ipv4Addr(127, 0, 0, 1), port: 34254 };
  43 ///     let mut socket = match UdpSocket::bind(addr) {
  44 ///         Ok(s) => s,
  45 ///         Err(e) => panic!("couldn't bind socket: {}", e),
  46 ///     };
  47 ///
  48 ///     let mut buf = [0; 10];
  49 ///     match socket.recv_from(&mut buf) {
  50 ///         Ok((amt, src)) => {
  51 ///             // Send a reply to the socket we received data from
  52 ///             let buf = buf.slice_to_mut(amt);
  53 ///             buf.reverse();
  54 ///             socket.send_to(buf, src);
  55 ///         }
  56 ///         Err(e) => println!("couldn't receive a datagram: {}", e)
  57 ///     }
  58 ///     drop(socket); // close the socket
  59 /// }
  60 /// ```
  61 pub struct UdpSocket {
  62     inner: UdpSocketImp,
  63 }
  64
  65 impl UdpSocket {
  66     /// Creates a UDP socket from the given address.
  67     ///
  68     /// Address type can be any implementor of `ToSocketAddr` trait. See its
  69     /// documentation for concrete examples.
  70     pub fn bind<A: ToSocketAddr>(addr: A) -> IoResult<UdpSocket> {
  71         super::with_addresses(addr, |addr| {
  72             UdpSocketImp::bind(addr).map(|s| UdpSocket { inner: s })
  73         })
  74     }
  75
  76     /// Receives data from the socket. On success, returns the number of bytes
  77     /// read and the address from whence the data came.
  78     pub fn recv_from(&mut self, buf: &mut [u8]) -> IoResult<(uint, SocketAddr)> {
  79         self.inner.recv_from(buf)
  80     }
  81
  82     /// Sends data on the socket to the given address. Returns nothing on
  83     /// success.
  84     ///
  85     /// Address type can be any implementer of `ToSocketAddr` trait. See its
  86     /// documentation for concrete examples.
  87     pub fn send_to<A: ToSocketAddr>(&mut self, buf: &[u8], addr: A) -> IoResult<()> {
  88         super::with_addresses(addr, |addr| self.inner.send_to(buf, addr))
  89     }
  90
  91     /// Creates a `UdpStream`, which allows use of the `Reader` and `Writer`
  92     /// traits to receive and send data from the same address. This transfers
  93     /// ownership of the socket to the stream.
  94     ///
  95     /// Note that this call does not perform any actual network communication,
  96     /// because UDP is a datagram protocol.
  97     #[deprecated = "`UdpStream` has been deprecated"]
  98     #[allow(deprecated)]
  99     pub fn connect(self, other: SocketAddr) -> UdpStream {
 100         UdpStream {
 101             socket: self,
 102             connected_to: other,
 103         }
 104     }
 105
 106     /// Returns the socket address that this socket was created from.
 107     pub fn socket_name(&mut self) -> IoResult<SocketAddr> {
 108         self.inner.socket_name()
 109     }
 110
 111     /// Joins a multicast IP address (becomes a member of it)
 112     #[experimental]
 113     pub fn join_multicast(&mut self, multi: IpAddr) -> IoResult<()> {
 114         self.inner.join_multicast(multi)
 115     }
 116
 117     /// Leaves a multicast IP address (drops membership from it)
 118     #[experimental]
 119     pub fn leave_multicast(&mut self, multi: IpAddr) -> IoResult<()> {
 120         self.inner.leave_multicast(multi)
 121     }
 122
 123     /// Set the multicast loop flag to the specified value
 124     ///
 125     /// This lets multicast packets loop back to local sockets (if enabled)
 126     #[experimental]
 127     pub fn set_multicast_loop(&mut self, on: bool) -> IoResult<()> {
 128         self.inner.set_multicast_loop(on)
 129     }
 130
 131     /// Sets the multicast TTL
 132     #[experimental]
 133     pub fn set_multicast_ttl(&mut self, ttl: int) -> IoResult<()> {
 134         self.inner.multicast_time_to_live(ttl)
 135     }
 136
 137     /// Sets this socket's TTL
 138     #[experimental]
 139     pub fn set_ttl(&mut self, ttl: int) -> IoResult<()> {
 140         self.inner.time_to_live(ttl)
 141     }
 142
 143     /// Sets the broadcast flag on or off
 144     #[experimental]
 145     pub fn set_broadcast(&mut self, broadcast: bool) -> IoResult<()> {
 146         self.inner.set_broadcast(broadcast)
 147     }
 148
 149     /// Sets the read/write timeout for this socket.
 150     ///
 151     /// For more information, see `TcpStream::set_timeout`
 152     #[experimental = "the timeout argument may change in type and value"]
 153     pub fn set_timeout(&mut self, timeout_ms: Option<u64>) {
 154         self.inner.set_timeout(timeout_ms)
 155     }
 156
 157     /// Sets the read timeout for this socket.
 158     ///
 159     /// For more information, see `TcpStream::set_timeout`
 160     #[experimental = "the timeout argument may change in type and value"]
 161     pub fn set_read_timeout(&mut self, timeout_ms: Option<u64>) {
 162         self.inner.set_read_timeout(timeout_ms)
 163     }
 164
 165     /// Sets the write timeout for this socket.
 166     ///
 167     /// For more information, see `TcpStream::set_timeout`
 168     #[experimental = "the timeout argument may change in type and value"]
 169     pub fn set_write_timeout(&mut self, timeout_ms: Option<u64>) {
 170         self.inner.set_write_timeout(timeout_ms)
 171     }
 172 }
 173
 174 impl Clone for UdpSocket {
 175     /// Creates a new handle to this UDP socket, allowing for simultaneous
 176     /// reads and writes of the socket.
 177     ///
 178     /// The underlying UDP socket will not be closed until all handles to the
 179     /// socket have been deallocated. Two concurrent reads will not receive
 180     /// the same data. Instead, the first read will receive the first packet
 181     /// received, and the second read will receive the second packet.
 182     fn clone(&self) -> UdpSocket {
 183         UdpSocket {
 184             inner: self.inner.clone(),
 185         }
 186     }
 187 }
 188
 189 impl sys_common::AsInner<UdpSocketImp> for UdpSocket {
 190     fn as_inner(&self) -> &UdpSocketImp {
 191         &self.inner
 192     }
 193 }
 194
 195 /// A type that allows convenient usage of a UDP stream connected to one
 196 /// address via the `Reader` and `Writer` traits.
 197 ///
 198 /// # Note
 199 ///
 200 /// This structure has been deprecated because `Reader` is a stream-oriented API but UDP
 201 /// is a packet-oriented protocol. Every `Reader` method will read a whole packet and
 202 /// throw all superfluous bytes away so that they are no longer available for further
 203 /// method calls.
 204 #[deprecated]
 205 pub struct UdpStream {
 206     socket: UdpSocket,
 207     connected_to: SocketAddr
 208 }
 209
 210 impl UdpStream {
 211     /// Allows access to the underlying UDP socket owned by this stream. This
 212     /// is useful to, for example, use the socket to send data to hosts other
 213     /// than the one that this stream is connected to.
 214     pub fn as_socket<T, F>(&mut self, f: F) -> T where
 215         F: FnOnce(&mut UdpSocket) -> T,
 216     {
 217         f(&mut self.socket)
 218     }
 219
 220     /// Consumes this UDP stream and returns out the underlying socket.
 221     pub fn disconnect(self) -> UdpSocket {
 222         self.socket
 223     }
 224 }
 225
 226 impl Reader for UdpStream {
 227     /// Returns the next non-empty message from the specified address.
 228     fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> {
 229         let peer = self.connected_to;
 230         self.as_socket(|sock| {
 231             loop {
 232                 let (nread, src) = try!(sock.recv_from(buf));
 233                 if nread > 0 && src == peer {
 234                     return Ok(nread);
 235                 }
 236             }
 237         })
 238     }
 239 }
 240
 241 impl Writer for UdpStream {
 242     fn write(&mut self, buf: &[u8]) -> IoResult<()> {
 243         let connected_to = self.connected_to;
 244         self.as_socket(|sock| sock.send_to(buf, connected_to))
 245     }
 246 }
 247
 248 #[cfg(test)]
 249 #[allow(experimental)]
 250 mod test {
 251     use prelude::v1::*;
 252
 253     use sync::mpsc::channel;
 254     use io::net::ip::*;
 255     use io::test::*;
 256     use io::{IoError, TimedOut, PermissionDenied, ShortWrite};
 257     use super::*;
 258     use thread::Thread;
 259
 260     // FIXME #11530 this fails on android because tests are run as root
 261     #[cfg_attr(any(windows, target_os = "android"), ignore)]
 262     #[test]
 263     fn bind_error() {
 264         let addr = SocketAddr { ip: Ipv4Addr(0, 0, 0, 0), port: 1 };
 265         match UdpSocket::bind(addr) {
 266             Ok(..) => panic!(),
 267             Err(e) => assert_eq!(e.kind, PermissionDenied),
 268         }
 269     }
 270
 271     #[test]
 272     fn socket_smoke_test_ip4() {
 273         let server_ip = next_test_ip4();
 274         let client_ip = next_test_ip4();
 275         let (tx1, rx1) = channel();
 276         let (tx2, rx2) = channel();
 277
 278         let _t = Thread::spawn(move|| {
 279             match UdpSocket::bind(client_ip) {
 280                 Ok(ref mut client) => {
 281                     rx1.recv().unwrap();
 282                     client.send_to(&[99], server_ip).unwrap()
 283                 }
 284                 Err(..) => panic!()
 285             }
 286             tx2.send(()).unwrap();
 287         });
 288
 289         match UdpSocket::bind(server_ip) {
 290             Ok(ref mut server) => {
 291                 tx1.send(()).unwrap();
 292                 let mut buf = [0];
 293                 match server.recv_from(&mut buf) {
 294                     Ok((nread, src)) => {
 295                         assert_eq!(nread, 1);
 296                         assert_eq!(buf[0], 99);
 297                         assert_eq!(src, client_ip);
 298                     }
 299                     Err(..) => panic!()
 300                 }
 301             }
 302             Err(..) => panic!()
 303         }
 304         rx2.recv().unwrap();
 305     }
 306
 307     #[test]
 308     fn socket_smoke_test_ip6() {
 309         let server_ip = next_test_ip6();
 310         let client_ip = next_test_ip6();
 311         let (tx, rx) = channel::<()>();
 312
 313         let _t = Thread::spawn(move|| {
 314             match UdpSocket::bind(client_ip) {
 315                 Ok(ref mut client) => {
 316                     rx.recv().unwrap();
 317                     client.send_to(&[99], server_ip).unwrap()
 318                 }
 319                 Err(..) => panic!()
 320             }
 321         });
 322
 323         match UdpSocket::bind(server_ip) {
 324             Ok(ref mut server) => {
 325                 tx.send(()).unwrap();
 326                 let mut buf = [0];
 327                 match server.recv_from(&mut buf) {
 328                     Ok((nread, src)) => {
 329                         assert_eq!(nread, 1);
 330                         assert_eq!(buf[0], 99);
 331                         assert_eq!(src, client_ip);
 332                     }
 333                     Err(..) => panic!()
 334                 }
 335             }
 336             Err(..) => panic!()
 337         }
 338     }
 339
 340     #[test]
 341     #[allow(deprecated)]
 342     fn stream_smoke_test_ip4() {
 343         let server_ip = next_test_ip4();
 344         let client_ip = next_test_ip4();
 345         let dummy_ip = next_test_ip4();
 346         let (tx1, rx1) = channel();
 347         let (tx2, rx2) = channel();
 348
 349         let _t = Thread::spawn(move|| {
 350             let send_as = |&:ip, val: &[u8]| {
 351                 match UdpSocket::bind(ip) {
 352                     Ok(client) => {
 353                         let client = box client;
 354                         let mut stream = client.connect(server_ip);
 355                         stream.write(val).unwrap();
 356                     }
 357                     Err(..) => panic!()
 358                 }
 359             };
 360             rx1.recv().unwrap();
 361             send_as(dummy_ip, &[98]);
 362             send_as(client_ip, &[99]);
 363             tx2.send(()).unwrap();
 364         });
 365
 366         match UdpSocket::bind(server_ip) {
 367             Ok(server) => {
 368                 let server = box server;
 369                 let mut stream = server.connect(client_ip);
 370                 tx1.send(()).unwrap();
 371                 let mut buf = [0];
 372                 match stream.read(&mut buf) {
 373                     Ok(nread) => {
 374                         assert_eq!(nread, 1);
 375                         assert_eq!(buf[0], 99);
 376                     }
 377                     Err(..) => panic!(),
 378                 }
 379             }
 380             Err(..) => panic!()
 381         }
 382         rx2.recv().unwrap();
 383     }
 384
 385     #[test]
 386     #[allow(deprecated)]
 387     fn stream_smoke_test_ip6() {
 388         let server_ip = next_test_ip6();
 389         let client_ip = next_test_ip6();
 390         let (tx1, rx1) = channel();
 391         let (tx2, rx2) = channel();
 392
 393         let _t = Thread::spawn(move|| {
 394             match UdpSocket::bind(client_ip) {
 395                 Ok(client) => {
 396                     let client = box client;
 397                     let mut stream = client.connect(server_ip);
 398                     rx1.recv().unwrap();
 399                     stream.write(&[99]).unwrap();
 400                 }
 401                 Err(..) => panic!()
 402             }
 403             tx2.send(()).unwrap();
 404         });
 405
 406         match UdpSocket::bind(server_ip) {
 407             Ok(server) => {
 408                 let server = box server;
 409                 let mut stream = server.connect(client_ip);
 410                 tx1.send(()).unwrap();
 411                 let mut buf = [0];
 412                 match stream.read(&mut buf) {
 413                     Ok(nread) => {
 414                         assert_eq!(nread, 1);
 415                         assert_eq!(buf[0], 99);
 416                     }
 417                     Err(..) => panic!()
 418                 }
 419             }
 420             Err(..) => panic!()
 421         }
 422         rx2.recv().unwrap();
 423     }
 424
 425     pub fn socket_name(addr: SocketAddr) {
 426         let server = UdpSocket::bind(addr);
 427
 428         assert!(server.is_ok());
 429         let mut server = server.unwrap();
 430
 431         // Make sure socket_name gives
 432         // us the socket we binded to.
 433         let so_name = server.socket_name();
 434         assert!(so_name.is_ok());
 435         assert_eq!(addr, so_name.unwrap());
 436     }
 437
 438     #[test]
 439     fn socket_name_ip4() {
 440         socket_name(next_test_ip4());
 441     }
 442
 443     #[test]
 444     fn socket_name_ip6() {
 445         socket_name(next_test_ip6());
 446     }
 447
 448     #[test]
 449     fn udp_clone_smoke() {
 450         let addr1 = next_test_ip4();
 451         let addr2 = next_test_ip4();
 452         let mut sock1 = UdpSocket::bind(addr1).unwrap();
 453         let sock2 = UdpSocket::bind(addr2).unwrap();
 454
 455         let _t = Thread::spawn(move|| {
 456             let mut sock2 = sock2;
 457             let mut buf = [0, 0];
 458             assert_eq!(sock2.recv_from(&mut buf), Ok((1, addr1)));
 459             assert_eq!(buf[0], 1);
 460             sock2.send_to(&[2], addr1).unwrap();
 461         });
 462
 463         let sock3 = sock1.clone();
 464
 465         let (tx1, rx1) = channel();
 466         let (tx2, rx2) = channel();
 467         let _t = Thread::spawn(move|| {
 468             let mut sock3 = sock3;
 469             rx1.recv().unwrap();
 470             sock3.send_to(&[1], addr2).unwrap();
 471             tx2.send(()).unwrap();
 472         });
 473         tx1.send(()).unwrap();
 474         let mut buf = [0, 0];
 475         assert_eq!(sock1.recv_from(&mut buf), Ok((1, addr2)));
 476         rx2.recv().unwrap();
 477     }
 478
 479     #[test]
 480     fn udp_clone_two_read() {
 481         let addr1 = next_test_ip4();
 482         let addr2 = next_test_ip4();
 483         let mut sock1 = UdpSocket::bind(addr1).unwrap();
 484         let sock2 = UdpSocket::bind(addr2).unwrap();
 485         let (tx1, rx) = channel();
 486         let tx2 = tx1.clone();
 487
 488         let _t = Thread::spawn(move|| {
 489             let mut sock2 = sock2;
 490             sock2.send_to(&[1], addr1).unwrap();
 491             rx.recv().unwrap();
 492             sock2.send_to(&[2], addr1).unwrap();
 493             rx.recv().unwrap();
 494         });
 495
 496         let sock3 = sock1.clone();
 497
 498         let (done, rx) = channel();
 499         let _t = Thread::spawn(move|| {
 500             let mut sock3 = sock3;
 501             let mut buf = [0, 0];
 502             sock3.recv_from(&mut buf).unwrap();
 503             tx2.send(()).unwrap();
 504             done.send(()).unwrap();
 505         });
 506         let mut buf = [0, 0];
 507         sock1.recv_from(&mut buf).unwrap();
 508         tx1.send(()).unwrap();
 509
 510         rx.recv().unwrap();
 511     }
 512
 513     #[test]
 514     fn udp_clone_two_write() {
 515         let addr1 = next_test_ip4();
 516         let addr2 = next_test_ip4();
 517         let mut sock1 = UdpSocket::bind(addr1).unwrap();
 518         let sock2 = UdpSocket::bind(addr2).unwrap();
 519
 520         let (tx, rx) = channel();
 521         let (serv_tx, serv_rx) = channel();
 522
 523         let _t = Thread::spawn(move|| {
 524             let mut sock2 = sock2;
 525             let mut buf = [0, 1];
 526
 527             rx.recv().unwrap();
 528             match sock2.recv_from(&mut buf) {
 529                 Ok(..) => {}
 530                 Err(e) => panic!("failed receive: {}", e),
 531             }
 532             serv_tx.send(()).unwrap();
 533         });
 534
 535         let sock3 = sock1.clone();
 536
 537         let (done, rx) = channel();
 538         let tx2 = tx.clone();
 539         let _t = Thread::spawn(move|| {
 540             let mut sock3 = sock3;
 541             match sock3.send_to(&[1], addr2) {
 542                 Ok(..) => { let _ = tx2.send(()); }
 543                 Err(..) => {}
 544             }
 545             done.send(()).unwrap();
 546         });
 547         match sock1.send_to(&[2], addr2) {
 548             Ok(..) => { let _ = tx.send(()); }
 549             Err(..) => {}
 550         }
 551         drop(tx);
 552
 553         rx.recv().unwrap();
 554         serv_rx.recv().unwrap();
 555     }
 556
 557     #[cfg(not(windows))] // FIXME #17553
 558     #[test]
 559     fn recv_from_timeout() {
 560         let addr1 = next_test_ip4();
 561         let addr2 = next_test_ip4();
 562         let mut a = UdpSocket::bind(addr1).unwrap();
 563         let a2 = UdpSocket::bind(addr2).unwrap();
 564
 565         let (tx, rx) = channel();
 566         let (tx2, rx2) = channel();
 567         let _t = Thread::spawn(move|| {
 568             let mut a = a2;
 569             assert_eq!(a.recv_from(&mut [0]), Ok((1, addr1)));
 570             assert_eq!(a.send_to(&[0], addr1), Ok(()));
 571             rx.recv().unwrap();
 572             assert_eq!(a.send_to(&[0], addr1), Ok(()));
 573
 574             tx2.send(()).unwrap();
 575         });
 576
 577         // Make sure that reads time out, but writes can continue
 578         a.set_read_timeout(Some(20));
 579         assert_eq!(a.recv_from(&mut [0]).err().unwrap().kind, TimedOut);
 580         assert_eq!(a.recv_from(&mut [0]).err().unwrap().kind, TimedOut);
 581         assert_eq!(a.send_to(&[0], addr2), Ok(()));
 582
 583         // Cloned handles should be able to block
 584         let mut a2 = a.clone();
 585         assert_eq!(a2.recv_from(&mut [0]), Ok((1, addr2)));
 586
 587         // Clearing the timeout should allow for receiving
 588         a.set_timeout(None);
 589         tx.send(()).unwrap();
 590         assert_eq!(a2.recv_from(&mut [0]), Ok((1, addr2)));
 591
 592         // Make sure the child didn't die
 593         rx2.recv().unwrap();
 594     }
 595
 596     #[test]
 597     fn send_to_timeout() {
 598         let addr1 = next_test_ip4();
 599         let addr2 = next_test_ip4();
 600         let mut a = UdpSocket::bind(addr1).unwrap();
 601         let _b = UdpSocket::bind(addr2).unwrap();
 602
 603         a.set_write_timeout(Some(1000));
 604         for _ in range(0u, 100) {
 605             match a.send_to(&[0;4*1024], addr2) {
 606                 Ok(()) | Err(IoError { kind: ShortWrite(..), .. }) => {},
 607                 Err(IoError { kind: TimedOut, .. }) => break,
 608                 Err(e) => panic!("other error: {}", e),
 609             }
 610         }
 611     }
 612 }