src/librustuv/net.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 use std::cast;
  12 use std::libc::{size_t, ssize_t, c_int, c_void, c_uint, c_char};
  13 use std::ptr;
  14 use std::rt::BlockedTask;
  15 use std::rt::io::IoError;
  16 use std::rt::io::net::ip::{Ipv4Addr, Ipv6Addr, SocketAddr, IpAddr};
  17 use std::rt::local::Local;
  18 use std::rt::rtio;
  19 use std::rt::sched::{Scheduler, SchedHandle};
  20 use std::rt::tube::Tube;
  21 use std::str;
  22 use std::vec;
  23
  24 use stream::StreamWatcher;
  25 use super::{Loop, Request, UvError, Buf, status_to_io_result,
  26             uv_error_to_io_error, UvHandle, slice_to_uv_buf};
  27 use uvio::HomingIO;
  28 use uvll;
  29
  30 ////////////////////////////////////////////////////////////////////////////////
  31 /// Generic functions related to dealing with sockaddr things
  32 ////////////////////////////////////////////////////////////////////////////////
  33
  34 pub enum UvSocketAddr {
  35     UvIpv4SocketAddr(*uvll::sockaddr_in),
  36     UvIpv6SocketAddr(*uvll::sockaddr_in6),
  37 }
  38
  39 pub fn sockaddr_to_UvSocketAddr(addr: *uvll::sockaddr) -> UvSocketAddr {
  40     unsafe {
  41         assert!((uvll::is_ip4_addr(addr) || uvll::is_ip6_addr(addr)));
  42         assert!(!(uvll::is_ip4_addr(addr) && uvll::is_ip6_addr(addr)));
  43         match addr {
  44             _ if uvll::is_ip4_addr(addr) =>
  45                 UvIpv4SocketAddr(addr as *uvll::sockaddr_in),
  46             _ if uvll::is_ip6_addr(addr) =>
  47                 UvIpv6SocketAddr(addr as *uvll::sockaddr_in6),
  48             _ => fail!(),
  49         }
  50     }
  51 }
  52
  53 fn socket_addr_as_uv_socket_addr<T>(addr: SocketAddr, f: &fn(UvSocketAddr) -> T) -> T {
  54     let malloc = match addr.ip {
  55         Ipv4Addr(*) => uvll::malloc_ip4_addr,
  56         Ipv6Addr(*) => uvll::malloc_ip6_addr,
  57     };
  58     let wrap = match addr.ip {
  59         Ipv4Addr(*) => UvIpv4SocketAddr,
  60         Ipv6Addr(*) => UvIpv6SocketAddr,
  61     };
  62     let free = match addr.ip {
  63         Ipv4Addr(*) => uvll::free_ip4_addr,
  64         Ipv6Addr(*) => uvll::free_ip6_addr,
  65     };
  66
  67     let addr = unsafe { malloc(addr.ip.to_str(), addr.port as int) };
  68     do (|| {
  69         f(wrap(addr))
  70     }).finally {
  71         unsafe { free(addr) };
  72     }
  73 }
  74
  75 fn uv_socket_addr_as_socket_addr<T>(addr: UvSocketAddr, f: &fn(SocketAddr) -> T) -> T {
  76     let ip_size = match addr {
  77         UvIpv4SocketAddr(*) => 4/*groups of*/ * 3/*digits separated by*/ + 3/*periods*/,
  78         UvIpv6SocketAddr(*) => 8/*groups of*/ * 4/*hex digits separated by*/ + 7 /*colons*/,
  79     };
  80     let ip_name = {
  81         let buf = vec::from_elem(ip_size + 1 /*null terminated*/, 0u8);
  82         unsafe {
  83             let buf_ptr = vec::raw::to_ptr(buf);
  84             match addr {
  85                 UvIpv4SocketAddr(addr) =>
  86                     uvll::uv_ip4_name(addr, buf_ptr as *c_char, ip_size as size_t),
  87                 UvIpv6SocketAddr(addr) =>
  88                     uvll::uv_ip6_name(addr, buf_ptr as *c_char, ip_size as size_t),
  89             }
  90         };
  91         buf
  92     };
  93     let ip_port = unsafe {
  94         let port = match addr {
  95             UvIpv4SocketAddr(addr) => uvll::ip4_port(addr),
  96             UvIpv6SocketAddr(addr) => uvll::ip6_port(addr),
  97         };
  98         port as u16
  99     };
 100     let ip_str = str::from_utf8_slice(ip_name).trim_right_chars(&'\x00');
 101     let ip_addr = FromStr::from_str(ip_str).unwrap();
 102
 103     // finally run the closure
 104     f(SocketAddr { ip: ip_addr, port: ip_port })
 105 }
 106
 107 pub fn uv_socket_addr_to_socket_addr(addr: UvSocketAddr) -> SocketAddr {
 108     use std::util;
 109     uv_socket_addr_as_socket_addr(addr, util::id)
 110 }
 111
 112 #[cfg(test)]
 113 #[test]
 114 fn test_ip4_conversion() {
 115     use std::rt;
 116     let ip4 = rt::test::next_test_ip4();
 117     assert_eq!(ip4, socket_addr_as_uv_socket_addr(ip4, uv_socket_addr_to_socket_addr));
 118 }
 119
 120 #[cfg(test)]
 121 #[test]
 122 fn test_ip6_conversion() {
 123     use std::rt;
 124     let ip6 = rt::test::next_test_ip6();
 125     assert_eq!(ip6, socket_addr_as_uv_socket_addr(ip6, uv_socket_addr_to_socket_addr));
 126 }
 127
 128 enum SocketNameKind {
 129     TcpPeer,
 130     Tcp,
 131     Udp
 132 }
 133
 134 fn socket_name(sk: SocketNameKind, handle: *c_void) -> Result<SocketAddr, IoError> {
 135     let getsockname = match sk {
 136         TcpPeer => uvll::tcp_getpeername,
 137         Tcp     => uvll::tcp_getsockname,
 138         Udp     => uvll::udp_getsockname,
 139     };
 140
 141     // Allocate a sockaddr_storage
 142     // since we don't know if it's ipv4 or ipv6
 143     let r_addr = unsafe { uvll::malloc_sockaddr_storage() };
 144
 145     let r = unsafe {
 146         getsockname(handle, r_addr as *uvll::sockaddr_storage)
 147     };
 148
 149     if r != 0 {
 150         return Err(uv_error_to_io_error(UvError(r)));
 151     }
 152
 153     let addr = unsafe {
 154         if uvll::is_ip6_addr(r_addr as *uvll::sockaddr) {
 155             uv_socket_addr_to_socket_addr(UvIpv6SocketAddr(r_addr as *uvll::sockaddr_in6))
 156         } else {
 157             uv_socket_addr_to_socket_addr(UvIpv4SocketAddr(r_addr as *uvll::sockaddr_in))
 158         }
 159     };
 160
 161     unsafe { uvll::free_sockaddr_storage(r_addr); }
 162
 163     Ok(addr)
 164
 165 }
 166
 167 ////////////////////////////////////////////////////////////////////////////////
 168 /// TCP implementation
 169 ////////////////////////////////////////////////////////////////////////////////
 170
 171 pub struct TcpWatcher {
 172     handle: *uvll::uv_tcp_t,
 173     stream: StreamWatcher,
 174     home: SchedHandle,
 175 }
 176
 177 pub struct TcpListener {
 178     home: SchedHandle,
 179     handle: *uvll::uv_pipe_t,
 180     priv closing_task: Option<BlockedTask>,
 181     priv outgoing: Tube<Result<~rtio::RtioTcpStream, IoError>>,
 182 }
 183
 184 pub struct TcpAcceptor {
 185     listener: ~TcpListener,
 186     priv incoming: Tube<Result<~rtio::RtioTcpStream, IoError>>,
 187 }
 188
 189 // TCP watchers (clients/streams)
 190
 191 impl TcpWatcher {
 192     pub fn new(loop_: &Loop) -> TcpWatcher {
 193         let handle = unsafe { uvll::malloc_handle(uvll::UV_TCP) };
 194         assert_eq!(unsafe {
 195             uvll::uv_tcp_init(loop_.handle, handle)
 196         }, 0);
 197         TcpWatcher {
 198             home: get_handle_to_current_scheduler!(),
 199             handle: handle,
 200             stream: StreamWatcher::new(handle),
 201         }
 202     }
 203
 204     pub fn connect(loop_: &mut Loop, address: SocketAddr)
 205         -> Result<TcpWatcher, UvError>
 206     {
 207         struct Ctx { status: c_int, task: Option<BlockedTask> }
 208
 209         let tcp = TcpWatcher::new(loop_);
 210         let ret = do socket_addr_as_uv_socket_addr(address) |addr| {
 211             let req = Request::new(uvll::UV_CONNECT);
 212             let result = match addr {
 213                 UvIpv4SocketAddr(addr) => unsafe {
 214                     uvll::tcp_connect(req.handle, tcp.handle, addr,
 215                                       connect_cb)
 216                 },
 217                 UvIpv6SocketAddr(addr) => unsafe {
 218                     uvll::tcp_connect6(req.handle, tcp.handle, addr,
 219                                        connect_cb)
 220                 },
 221             };
 222             match result {
 223                 0 => {
 224                     let mut cx = Ctx { status: 0, task: None };
 225                     req.set_data(&cx);
 226                     req.defuse();
 227                     let scheduler: ~Scheduler = Local::take();
 228                     do scheduler.deschedule_running_task_and_then |_, task| {
 229                         cx.task = Some(task);
 230                     }
 231                     match cx.status {
 232                         0 => Ok(()),
 233                         n => Err(UvError(n)),
 234                     }
 235                 }
 236                 n => Err(UvError(n))
 237             }
 238         };
 239
 240         return match ret {
 241             Ok(()) => Ok(tcp),
 242             Err(e) => Err(e),
 243         };
 244
 245         extern fn connect_cb(req: *uvll::uv_connect_t, status: c_int) {
 246             let req = Request::wrap(req);
 247             if status == uvll::ECANCELED { return }
 248             let cx: &mut Ctx = unsafe { cast::transmute(req.get_data()) };
 249             cx.status = status;
 250             let scheduler: ~Scheduler = Local::take();
 251             scheduler.resume_blocked_task_immediately(cx.task.take_unwrap());
 252         }
 253     }
 254 }
 255
 256 impl HomingIO for TcpWatcher {
 257     fn home<'r>(&'r mut self) -> &'r mut SchedHandle { &mut self.home }
 258 }
 259
 260 impl rtio::RtioSocket for TcpWatcher {
 261     fn socket_name(&mut self) -> Result<SocketAddr, IoError> {
 262         let _m = self.fire_missiles();
 263         socket_name(Tcp, self.handle)
 264     }
 265 }
 266
 267 impl rtio::RtioTcpStream for TcpWatcher {
 268     fn read(&mut self, buf: &mut [u8]) -> Result<uint, IoError> {
 269         let _m = self.fire_missiles();
 270         self.stream.read(buf).map_err(uv_error_to_io_error)
 271     }
 272
 273     fn write(&mut self, buf: &[u8]) -> Result<(), IoError> {
 274         let _m = self.fire_missiles();
 275         self.stream.write(buf).map_err(uv_error_to_io_error)
 276     }
 277
 278     fn peer_name(&mut self) -> Result<SocketAddr, IoError> {
 279         let _m = self.fire_missiles();
 280         socket_name(TcpPeer, self.handle)
 281     }
 282
 283     fn control_congestion(&mut self) -> Result<(), IoError> {
 284         let _m = self.fire_missiles();
 285         status_to_io_result(unsafe {
 286             uvll::uv_tcp_nodelay(self.handle, 0 as c_int)
 287         })
 288     }
 289
 290     fn nodelay(&mut self) -> Result<(), IoError> {
 291         let _m = self.fire_missiles();
 292         status_to_io_result(unsafe {
 293             uvll::uv_tcp_nodelay(self.handle, 1 as c_int)
 294         })
 295     }
 296
 297     fn keepalive(&mut self, delay_in_seconds: uint) -> Result<(), IoError> {
 298         let _m = self.fire_missiles();
 299         status_to_io_result(unsafe {
 300             uvll::uv_tcp_keepalive(self.handle, 1 as c_int,
 301                                    delay_in_seconds as c_uint)
 302         })
 303     }
 304
 305     fn letdie(&mut self) -> Result<(), IoError> {
 306         let _m = self.fire_missiles();
 307         status_to_io_result(unsafe {
 308             uvll::uv_tcp_keepalive(self.handle, 0 as c_int, 0 as c_uint)
 309         })
 310     }
 311 }
 312
 313 impl Drop for TcpWatcher {
 314     fn drop(&mut self) {
 315         let _m = self.fire_missiles();
 316         self.stream.close();
 317     }
 318 }
 319
 320 // TCP listeners (unbound servers)
 321
 322 impl TcpListener {
 323     pub fn bind(loop_: &mut Loop, address: SocketAddr)
 324         -> Result<~TcpListener, UvError>
 325     {
 326         let handle = unsafe { uvll::malloc_handle(uvll::UV_TCP) };
 327         assert_eq!(unsafe {
 328             uvll::uv_tcp_init(loop_.handle, handle)
 329         }, 0);
 330         let l = ~TcpListener {
 331             home: get_handle_to_current_scheduler!(),
 332             handle: handle,
 333             closing_task: None,
 334             outgoing: Tube::new(),
 335         };
 336         let res = socket_addr_as_uv_socket_addr(address, |addr| unsafe {
 337             match addr {
 338                 UvIpv4SocketAddr(addr) => uvll::tcp_bind(l.handle, addr),
 339                 UvIpv6SocketAddr(addr) => uvll::tcp_bind6(l.handle, addr),
 340             }
 341         });
 342         match res {
 343             0 => Ok(l.install()),
 344             n => Err(UvError(n))
 345         }
 346     }
 347 }
 348
 349 impl HomingIO for TcpListener {
 350     fn home<'r>(&'r mut self) -> &'r mut SchedHandle { &mut self.home }
 351 }
 352
 353 impl UvHandle<uvll::uv_tcp_t> for TcpListener {
 354     fn uv_handle(&self) -> *uvll::uv_tcp_t { self.handle }
 355 }
 356
 357 impl rtio::RtioSocket for TcpListener {
 358     fn socket_name(&mut self) -> Result<SocketAddr, IoError> {
 359         let _m = self.fire_missiles();
 360         socket_name(Tcp, self.handle)
 361     }
 362 }
 363
 364 impl rtio::RtioTcpListener for TcpListener {
 365     fn listen(mut ~self) -> Result<~rtio::RtioTcpAcceptor, IoError> {
 366         // create the acceptor object from ourselves
 367         let incoming = self.outgoing.clone();
 368         let mut acceptor = ~TcpAcceptor {
 369             listener: self,
 370             incoming: incoming,
 371         };
 372
 373         let _m = acceptor.fire_missiles();
 374         // XXX: the 128 backlog should be configurable
 375         match unsafe { uvll::uv_listen(acceptor.listener.handle, 128, listen_cb) } {
 376             0 => Ok(acceptor as ~rtio::RtioTcpAcceptor),
 377             n => Err(uv_error_to_io_error(UvError(n))),
 378         }
 379     }
 380 }
 381
 382 extern fn listen_cb(server: *uvll::uv_stream_t, status: c_int) {
 383     let msg = match status {
 384         0 => {
 385             let loop_ = Loop::wrap(unsafe {
 386                 uvll::get_loop_for_uv_handle(server)
 387             });
 388             let client = TcpWatcher::new(&loop_);
 389             assert_eq!(unsafe { uvll::uv_accept(server, client.handle) }, 0);
 390             Ok(~client as ~rtio::RtioTcpStream)
 391         }
 392         uvll::ECANCELED => return,
 393         n => Err(uv_error_to_io_error(UvError(n)))
 394     };
 395
 396     let tcp: &mut TcpListener = unsafe { UvHandle::from_uv_handle(&server) };
 397     tcp.outgoing.send(msg);
 398 }
 399
 400 impl Drop for TcpListener {
 401     fn drop(&mut self) {
 402         let (_m, sched) = self.fire_missiles_sched();
 403
 404         do sched.deschedule_running_task_and_then |_, task| {
 405             self.closing_task = Some(task);
 406             unsafe { uvll::uv_close(self.handle, listener_close_cb) }
 407         }
 408     }
 409 }
 410
 411 extern fn listener_close_cb(handle: *uvll::uv_handle_t) {
 412     let tcp: &mut TcpListener = unsafe { UvHandle::from_uv_handle(&handle) };
 413     unsafe { uvll::free_handle(handle) }
 414
 415     let sched: ~Scheduler = Local::take();
 416     sched.resume_blocked_task_immediately(tcp.closing_task.take_unwrap());
 417 }
 418
 419 // TCP acceptors (bound servers)
 420
 421 impl HomingIO for TcpAcceptor {
 422     fn home<'r>(&'r mut self) -> &'r mut SchedHandle { self.listener.home() }
 423 }
 424
 425 impl rtio::RtioSocket for TcpAcceptor {
 426     fn socket_name(&mut self) -> Result<SocketAddr, IoError> {
 427         let _m = self.fire_missiles();
 428         socket_name(Tcp, self.listener.handle)
 429     }
 430 }
 431
 432 impl rtio::RtioTcpAcceptor for TcpAcceptor {
 433     fn accept(&mut self) -> Result<~rtio::RtioTcpStream, IoError> {
 434         let _m = self.fire_missiles();
 435         self.incoming.recv()
 436     }
 437
 438     fn accept_simultaneously(&mut self) -> Result<(), IoError> {
 439         let _m = self.fire_missiles();
 440         status_to_io_result(unsafe {
 441             uvll::uv_tcp_simultaneous_accepts(self.listener.handle, 1)
 442         })
 443     }
 444
 445     fn dont_accept_simultaneously(&mut self) -> Result<(), IoError> {
 446         let _m = self.fire_missiles();
 447         status_to_io_result(unsafe {
 448             uvll::uv_tcp_simultaneous_accepts(self.listener.handle, 0)
 449         })
 450     }
 451 }
 452
 453 ////////////////////////////////////////////////////////////////////////////////
 454 /// UDP implementation
 455 ////////////////////////////////////////////////////////////////////////////////
 456
 457 pub struct UdpWatcher {
 458     handle: *uvll::uv_udp_t,
 459     home: SchedHandle,
 460 }
 461
 462 impl UdpWatcher {
 463     pub fn bind(loop_: &Loop, address: SocketAddr)
 464         -> Result<UdpWatcher, UvError>
 465     {
 466         let udp = UdpWatcher {
 467             handle: unsafe { uvll::malloc_handle(uvll::UV_UDP) },
 468             home: get_handle_to_current_scheduler!(),
 469         };
 470         assert_eq!(unsafe {
 471             uvll::uv_udp_init(loop_.handle, udp.handle)
 472         }, 0);
 473         let result = socket_addr_as_uv_socket_addr(address, |addr| unsafe {
 474             match addr {
 475                 UvIpv4SocketAddr(addr) => uvll::udp_bind(udp.handle, addr, 0u32),
 476                 UvIpv6SocketAddr(addr) => uvll::udp_bind6(udp.handle, addr, 0u32),
 477             }
 478         });
 479         match result {
 480             0 => Ok(udp),
 481             n => Err(UvError(n)),
 482         }
 483     }
 484 }
 485
 486 impl HomingIO for UdpWatcher {
 487     fn home<'r>(&'r mut self) -> &'r mut SchedHandle { &mut self.home }
 488 }
 489
 490 impl rtio::RtioSocket for UdpWatcher {
 491     fn socket_name(&mut self) -> Result<SocketAddr, IoError> {
 492         let _m = self.fire_missiles();
 493         socket_name(Udp, self.handle)
 494     }
 495 }
 496
 497 impl rtio::RtioUdpSocket for UdpWatcher {
 498     fn recvfrom(&mut self, buf: &mut [u8])
 499         -> Result<(uint, SocketAddr), IoError>
 500     {
 501         struct Ctx {
 502             task: Option<BlockedTask>,
 503             buf: Option<Buf>,
 504             result: Option<(ssize_t, SocketAddr)>,
 505         }
 506         let _m = self.fire_missiles();
 507
 508         return match unsafe {
 509             uvll::uv_udp_recv_start(self.handle, alloc_cb, recv_cb)
 510         } {
 511             0 => {
 512                 let mut cx = Ctx {
 513                     task: None,
 514                     buf: Some(slice_to_uv_buf(buf)),
 515                     result: None,
 516                 };
 517                 unsafe { uvll::set_data_for_uv_handle(self.handle, &cx) }
 518                 let scheduler: ~Scheduler = Local::take();
 519                 do scheduler.deschedule_running_task_and_then |_, task| {
 520                     cx.task = Some(task);
 521                 }
 522                 match cx.result.take_unwrap() {
 523                     (n, _) if n < 0 =>
 524                         Err(uv_error_to_io_error(UvError(n as c_int))),
 525                     (n, addr) => Ok((n as uint, addr))
 526                 }
 527             }
 528             n => Err(uv_error_to_io_error(UvError(n)))
 529         };
 530
 531         extern fn alloc_cb(handle: *uvll::uv_udp_t,
 532                            _suggested_size: size_t) -> Buf {
 533             let cx: &mut Ctx = unsafe {
 534                 cast::transmute(uvll::get_data_for_uv_handle(handle))
 535             };
 536             cx.buf.take().expect("alloc_cb called more than once")
 537         }
 538
 539         extern fn recv_cb(handle: *uvll::uv_udp_t, nread: ssize_t, _buf: Buf,
 540                           addr: *uvll::sockaddr, _flags: c_uint) {
 541
 542             // When there's no data to read the recv callback can be a no-op.
 543             // This can happen if read returns EAGAIN/EWOULDBLOCK. By ignoring
 544             // this we just drop back to kqueue and wait for the next callback.
 545             if nread == 0 { return }
 546             if nread == uvll::ECANCELED as ssize_t { return }
 547
 548             unsafe {
 549                 assert_eq!(uvll::uv_udp_recv_stop(handle), 0)
 550             }
 551
 552             let cx: &mut Ctx = unsafe {
 553                 cast::transmute(uvll::get_data_for_uv_handle(handle))
 554             };
 555             let addr = sockaddr_to_UvSocketAddr(addr);
 556             let addr = uv_socket_addr_to_socket_addr(addr);
 557             cx.result = Some((nread, addr));
 558
 559             let sched: ~Scheduler = Local::take();
 560             sched.resume_blocked_task_immediately(cx.task.take_unwrap());
 561         }
 562     }
 563
 564     fn sendto(&mut self, buf: &[u8], dst: SocketAddr) -> Result<(), IoError> {
 565         struct Ctx { task: Option<BlockedTask>, result: c_int }
 566
 567         let _m = self.fire_missiles();
 568
 569         let req = Request::new(uvll::UV_UDP_SEND);
 570         let buf = slice_to_uv_buf(buf);
 571         let result = socket_addr_as_uv_socket_addr(dst, |dst| unsafe {
 572             match dst {
 573                 UvIpv4SocketAddr(dst) =>
 574                     uvll::udp_send(req.handle, self.handle, [buf], dst, send_cb),
 575                 UvIpv6SocketAddr(dst) =>
 576                     uvll::udp_send6(req.handle, self.handle, [buf], dst, send_cb),
 577             }
 578         });
 579
 580         return match result {
 581             0 => {
 582                 let mut cx = Ctx { task: None, result: 0 };
 583                 req.set_data(&cx);
 584                 req.defuse();
 585
 586                 let sched: ~Scheduler = Local::take();
 587                 do sched.deschedule_running_task_and_then |_, task| {
 588                     cx.task = Some(task);
 589                 }
 590
 591                 match cx.result {
 592                     0 => Ok(()),
 593                     n => Err(uv_error_to_io_error(UvError(n)))
 594                 }
 595             }
 596             n => Err(uv_error_to_io_error(UvError(n)))
 597         };
 598
 599         extern fn send_cb(req: *uvll::uv_udp_send_t, status: c_int) {
 600             let req = Request::wrap(req);
 601             let cx: &mut Ctx = unsafe { cast::transmute(req.get_data()) };
 602             cx.result = status;
 603
 604             let sched: ~Scheduler = Local::take();
 605             sched.resume_blocked_task_immediately(cx.task.take_unwrap());
 606         }
 607     }
 608
 609     fn join_multicast(&mut self, multi: IpAddr) -> Result<(), IoError> {
 610         let _m = self.fire_missiles();
 611         status_to_io_result(unsafe {
 612             do multi.to_str().with_c_str |m_addr| {
 613                 uvll::uv_udp_set_membership(self.handle,
 614                                             m_addr, ptr::null(),
 615                                             uvll::UV_JOIN_GROUP)
 616             }
 617         })
 618     }
 619
 620     fn leave_multicast(&mut self, multi: IpAddr) -> Result<(), IoError> {
 621         let _m = self.fire_missiles();
 622         status_to_io_result(unsafe {
 623             do multi.to_str().with_c_str |m_addr| {
 624                 uvll::uv_udp_set_membership(self.handle,
 625                                             m_addr, ptr::null(),
 626                                             uvll::UV_LEAVE_GROUP)
 627             }
 628         })
 629     }
 630
 631     fn loop_multicast_locally(&mut self) -> Result<(), IoError> {
 632         let _m = self.fire_missiles();
 633         status_to_io_result(unsafe {
 634             uvll::uv_udp_set_multicast_loop(self.handle,
 635                                             1 as c_int)
 636         })
 637     }
 638
 639     fn dont_loop_multicast_locally(&mut self) -> Result<(), IoError> {
 640         let _m = self.fire_missiles();
 641         status_to_io_result(unsafe {
 642             uvll::uv_udp_set_multicast_loop(self.handle,
 643                                             0 as c_int)
 644         })
 645     }
 646
 647     fn multicast_time_to_live(&mut self, ttl: int) -> Result<(), IoError> {
 648         let _m = self.fire_missiles();
 649         status_to_io_result(unsafe {
 650             uvll::uv_udp_set_multicast_ttl(self.handle,
 651                                            ttl as c_int)
 652         })
 653     }
 654
 655     fn time_to_live(&mut self, ttl: int) -> Result<(), IoError> {
 656         let _m = self.fire_missiles();
 657         status_to_io_result(unsafe {
 658             uvll::uv_udp_set_ttl(self.handle, ttl as c_int)
 659         })
 660     }
 661
 662     fn hear_broadcasts(&mut self) -> Result<(), IoError> {
 663         let _m = self.fire_missiles();
 664         status_to_io_result(unsafe {
 665             uvll::uv_udp_set_broadcast(self.handle,
 666                                        1 as c_int)
 667         })
 668     }
 669
 670     fn ignore_broadcasts(&mut self) -> Result<(), IoError> {
 671         let _m = self.fire_missiles();
 672         status_to_io_result(unsafe {
 673             uvll::uv_udp_set_broadcast(self.handle,
 674                                        0 as c_int)
 675         })
 676     }
 677 }
 678
 679 impl Drop for UdpWatcher {
 680     fn drop(&mut self) {
 681         // Send ourselves home to close this handle (blocking while doing so).
 682         let (_m, sched) = self.fire_missiles_sched();
 683         let mut slot = None;
 684         unsafe {
 685             uvll::set_data_for_uv_handle(self.handle, &slot);
 686             uvll::uv_close(self.handle, close_cb);
 687         }
 688         do sched.deschedule_running_task_and_then |_, task| {
 689             slot = Some(task);
 690         }
 691
 692         extern fn close_cb(handle: *uvll::uv_handle_t) {
 693             let slot: &mut Option<BlockedTask> = unsafe {
 694                 cast::transmute(uvll::get_data_for_uv_handle(handle))
 695             };
 696             let sched: ~Scheduler = Local::take();
 697             sched.resume_blocked_task_immediately(slot.take_unwrap());
 698         }
 699     }
 700 }
 701
 702 ////////////////////////////////////////////////////////////////////////////////
 703 /// UV request support
 704 ////////////////////////////////////////////////////////////////////////////////
 705
 706 #[cfg(test)]
 707 mod test {
 708     use super::*;
 709     use std::util::ignore;
 710     use std::cell::Cell;
 711     use std::vec;
 712     use std::unstable::run_in_bare_thread;
 713     use std::rt::thread::Thread;
 714     use std::rt::test::*;
 715     use super::super::{Loop, AllocCallback};
 716     use super::super::{vec_from_uv_buf, vec_to_uv_buf, slice_to_uv_buf};
 717
 718     #[test]
 719     fn connect_close_ip4() {
 720         do run_in_bare_thread() {
 721             let mut loop_ = Loop::new();
 722             let mut tcp_watcher = { TcpWatcher::new(&mut loop_) };
 723             // Connect to a port where nobody is listening
 724             let addr = next_test_ip4();
 725             do tcp_watcher.connect(addr) |stream_watcher, status| {
 726                 uvdebug!("tcp_watcher.connect!");
 727                 assert!(status.is_some());
 728                 assert_eq!(status.unwrap().name(), ~"ECONNREFUSED");
 729                 stream_watcher.close(||());
 730             }
 731             loop_.run();
 732             loop_.close();
 733         }
 734     }
 735
 736     #[test]
 737     fn connect_close_ip6() {
 738         do run_in_bare_thread() {
 739             let mut loop_ = Loop::new();
 740             let mut tcp_watcher = { TcpWatcher::new(&mut loop_) };
 741             // Connect to a port where nobody is listening
 742             let addr = next_test_ip6();
 743             do tcp_watcher.connect(addr) |stream_watcher, status| {
 744                 uvdebug!("tcp_watcher.connect!");
 745                 assert!(status.is_some());
 746                 assert_eq!(status.unwrap().name(), ~"ECONNREFUSED");
 747                 stream_watcher.close(||());
 748             }
 749             loop_.run();
 750             loop_.close();
 751         }
 752     }
 753
 754     #[test]
 755     fn udp_bind_close_ip4() {
 756         do run_in_bare_thread() {
 757             let mut loop_ = Loop::new();
 758             let mut udp_watcher = { UdpWatcher::new(&mut loop_) };
 759             let addr = next_test_ip4();
 760             udp_watcher.bind(addr);
 761             udp_watcher.close(||());
 762             loop_.run();
 763             loop_.close();
 764         }
 765     }
 766
 767     #[test]
 768     fn udp_bind_close_ip6() {
 769         do run_in_bare_thread() {
 770             let mut loop_ = Loop::new();
 771             let mut udp_watcher = { UdpWatcher::new(&mut loop_) };
 772             let addr = next_test_ip6();
 773             udp_watcher.bind(addr);
 774             udp_watcher.close(||());
 775             loop_.run();
 776             loop_.close();
 777         }
 778     }
 779
 780     #[test]
 781     fn listen_ip4() {
 782         do run_in_bare_thread() {
 783             static MAX: int = 10;
 784             let mut loop_ = Loop::new();
 785             let mut server_tcp_watcher = { TcpWatcher::new(&mut loop_) };
 786             let addr = next_test_ip4();
 787             server_tcp_watcher.bind(addr);
 788             let loop_ = loop_;
 789             uvdebug!("listening");
 790             let mut stream = server_tcp_watcher.as_stream();
 791             let res = do stream.listen |mut server_stream_watcher, status| {
 792                 uvdebug!("listened!");
 793                 assert!(status.is_none());
 794                 let mut loop_ = loop_;
 795                 let client_tcp_watcher = TcpWatcher::new(&mut loop_);
 796                 let mut client_tcp_watcher = client_tcp_watcher.as_stream();
 797                 server_stream_watcher.accept(client_tcp_watcher);
 798                 let count_cell = Cell::new(0);
 799                 let server_stream_watcher = server_stream_watcher;
 800                 uvdebug!("starting read");
 801                 let alloc: AllocCallback = |size| {
 802                     vec_to_uv_buf(vec::from_elem(size, 0u8))
 803                 };
 804                 do client_tcp_watcher.read_start(alloc) |stream_watcher, nread, buf, status| {
 805
 806                     uvdebug!("i'm reading!");
 807                     let buf = vec_from_uv_buf(buf);
 808                     let mut count = count_cell.take();
 809                     if status.is_none() {
 810                         uvdebug!("got {} bytes", nread);
 811                         let buf = buf.unwrap();
 812                         for byte in buf.slice(0, nread as uint).iter() {
 813                             assert!(*byte == count as u8);
 814                             uvdebug!("{}", *byte as uint);
 815                             count += 1;
 816                         }
 817                     } else {
 818                         assert_eq!(count, MAX);
 819                         do stream_watcher.close {
 820                             server_stream_watcher.close(||());
 821                         }
 822                     }
 823                     count_cell.put_back(count);
 824                 }
 825             };
 826
 827             assert!(res.is_ok());
 828
 829             let client_thread = do Thread::start {
 830                 uvdebug!("starting client thread");
 831                 let mut loop_ = Loop::new();
 832                 let mut tcp_watcher = { TcpWatcher::new(&mut loop_) };
 833                 do tcp_watcher.connect(addr) |mut stream_watcher, status| {
 834                     uvdebug!("connecting");
 835                     assert!(status.is_none());
 836                     let msg = ~[0, 1, 2, 3, 4, 5, 6 ,7 ,8, 9];
 837                     let buf = slice_to_uv_buf(msg);
 838                     let msg_cell = Cell::new(msg);
 839                     do stream_watcher.write(buf) |stream_watcher, status| {
 840                         uvdebug!("writing");
 841                         assert!(status.is_none());
 842                         let msg_cell = Cell::new(msg_cell.take());
 843                         stream_watcher.close(||ignore(msg_cell.take()));
 844                     }
 845                 }
 846                 loop_.run();
 847                 loop_.close();
 848             };
 849
 850             let mut loop_ = loop_;
 851             loop_.run();
 852             loop_.close();
 853             client_thread.join();
 854         };
 855     }
 856
 857     #[test]
 858     fn listen_ip6() {
 859         do run_in_bare_thread() {
 860             static MAX: int = 10;
 861             let mut loop_ = Loop::new();
 862             let mut server_tcp_watcher = { TcpWatcher::new(&mut loop_) };
 863             let addr = next_test_ip6();
 864             server_tcp_watcher.bind(addr);
 865             let loop_ = loop_;
 866             uvdebug!("listening");
 867             let mut stream = server_tcp_watcher.as_stream();
 868             let res = do stream.listen |mut server_stream_watcher, status| {
 869                 uvdebug!("listened!");
 870                 assert!(status.is_none());
 871                 let mut loop_ = loop_;
 872                 let client_tcp_watcher = TcpWatcher::new(&mut loop_);
 873                 let mut client_tcp_watcher = client_tcp_watcher.as_stream();
 874                 server_stream_watcher.accept(client_tcp_watcher);
 875                 let count_cell = Cell::new(0);
 876                 let server_stream_watcher = server_stream_watcher;
 877                 uvdebug!("starting read");
 878                 let alloc: AllocCallback = |size| {
 879                     vec_to_uv_buf(vec::from_elem(size, 0u8))
 880                 };
 881                 do client_tcp_watcher.read_start(alloc)
 882                     |stream_watcher, nread, buf, status| {
 883
 884                     uvdebug!("i'm reading!");
 885                     let buf = vec_from_uv_buf(buf);
 886                     let mut count = count_cell.take();
 887                     if status.is_none() {
 888                         uvdebug!("got {} bytes", nread);
 889                         let buf = buf.unwrap();
 890                         let r = buf.slice(0, nread as uint);
 891                         for byte in r.iter() {
 892                             assert!(*byte == count as u8);
 893                             uvdebug!("{}", *byte as uint);
 894                             count += 1;
 895                         }
 896                     } else {
 897                         assert_eq!(count, MAX);
 898                         do stream_watcher.close {
 899                             server_stream_watcher.close(||());
 900                         }
 901                     }
 902                     count_cell.put_back(count);
 903                 }
 904             };
 905             assert!(res.is_ok());
 906
 907             let client_thread = do Thread::start {
 908                 uvdebug!("starting client thread");
 909                 let mut loop_ = Loop::new();
 910                 let mut tcp_watcher = { TcpWatcher::new(&mut loop_) };
 911                 do tcp_watcher.connect(addr) |mut stream_watcher, status| {
 912                     uvdebug!("connecting");
 913                     assert!(status.is_none());
 914                     let msg = ~[0, 1, 2, 3, 4, 5, 6 ,7 ,8, 9];
 915                     let buf = slice_to_uv_buf(msg);
 916                     let msg_cell = Cell::new(msg);
 917                     do stream_watcher.write(buf) |stream_watcher, status| {
 918                         uvdebug!("writing");
 919                         assert!(status.is_none());
 920                         let msg_cell = Cell::new(msg_cell.take());
 921                         stream_watcher.close(||ignore(msg_cell.take()));
 922                     }
 923                 }
 924                 loop_.run();
 925                 loop_.close();
 926             };
 927
 928             let mut loop_ = loop_;
 929             loop_.run();
 930             loop_.close();
 931             client_thread.join();
 932         }
 933     }
 934
 935     #[test]
 936     fn udp_recv_ip4() {
 937         do run_in_bare_thread() {
 938             static MAX: int = 10;
 939             let mut loop_ = Loop::new();
 940             let server_addr = next_test_ip4();
 941             let client_addr = next_test_ip4();
 942
 943             let mut server = UdpWatcher::new(&loop_);
 944             assert!(server.bind(server_addr).is_ok());
 945
 946             uvdebug!("starting read");
 947             let alloc: AllocCallback = |size| {
 948                 vec_to_uv_buf(vec::from_elem(size, 0u8))
 949             };
 950
 951             do server.recv_start(alloc) |mut server, nread, buf, src, flags, status| {
 952                 server.recv_stop();
 953                 uvdebug!("i'm reading!");
 954                 assert!(status.is_none());
 955                 assert_eq!(flags, 0);
 956                 assert_eq!(src, client_addr);
 957
 958                 let buf = vec_from_uv_buf(buf);
 959                 let mut count = 0;
 960                 uvdebug!("got {} bytes", nread);
 961
 962                 let buf = buf.unwrap();
 963                 for &byte in buf.slice(0, nread as uint).iter() {
 964                     assert!(byte == count as u8);
 965                     uvdebug!("{}", byte as uint);
 966                     count += 1;
 967                 }
 968                 assert_eq!(count, MAX);
 969
 970                 server.close(||{});
 971             }
 972
 973             let thread = do Thread::start {
 974                 let mut loop_ = Loop::new();
 975                 let mut client = UdpWatcher::new(&loop_);
 976                 assert!(client.bind(client_addr).is_ok());
 977                 let msg = ~[0, 1, 2, 3, 4, 5, 6, 7, 8, 9];
 978                 let buf = slice_to_uv_buf(msg);
 979                 do client.send(buf, server_addr) |client, status| {
 980                     uvdebug!("writing");
 981                     assert!(status.is_none());
 982                     client.close(||{});
 983                 }
 984
 985                 loop_.run();
 986                 loop_.close();
 987             };
 988
 989             loop_.run();
 990             loop_.close();
 991             thread.join();
 992         }
 993     }
 994
 995     #[test]
 996     fn udp_recv_ip6() {
 997         do run_in_bare_thread() {
 998             static MAX: int = 10;
 999             let mut loop_ = Loop::new();
1000             let server_addr = next_test_ip6();
1001             let client_addr = next_test_ip6();
1002
1003             let mut server = UdpWatcher::new(&loop_);
1004             assert!(server.bind(server_addr).is_ok());
1005
1006             uvdebug!("starting read");
1007             let alloc: AllocCallback = |size| {
1008                 vec_to_uv_buf(vec::from_elem(size, 0u8))
1009             };
1010
1011             do server.recv_start(alloc) |mut server, nread, buf, src, flags, status| {
1012                 server.recv_stop();
1013                 uvdebug!("i'm reading!");
1014                 assert!(status.is_none());
1015                 assert_eq!(flags, 0);
1016                 assert_eq!(src, client_addr);
1017
1018                 let buf = vec_from_uv_buf(buf);
1019                 let mut count = 0;
1020                 uvdebug!("got {} bytes", nread);
1021
1022                 let buf = buf.unwrap();
1023                 for &byte in buf.slice(0, nread as uint).iter() {
1024                     assert!(byte == count as u8);
1025                     uvdebug!("{}", byte as uint);
1026                     count += 1;
1027                 }
1028                 assert_eq!(count, MAX);
1029
1030                 server.close(||{});
1031             }
1032
1033             let thread = do Thread::start {
1034                 let mut loop_ = Loop::new();
1035                 let mut client = UdpWatcher::new(&loop_);
1036                 assert!(client.bind(client_addr).is_ok());
1037                 let msg = ~[0, 1, 2, 3, 4, 5, 6, 7, 8, 9];
1038                 let buf = slice_to_uv_buf(msg);
1039                 do client.send(buf, server_addr) |client, status| {
1040                     uvdebug!("writing");
1041                     assert!(status.is_none());
1042                     client.close(||{});
1043                 }
1044
1045                 loop_.run();
1046                 loop_.close();
1047             };
1048
1049             loop_.run();
1050             loop_.close();
1051             thread.join();
1052         }
1053     }
1054 }