src/libstd/io/net/tcp.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 //! TCP network connections
  12 //!
  13 //! This module contains the ability to open a TCP stream to a socket address,
  14 //! as well as creating a socket server to accept incoming connections. The
  15 //! destination and binding addresses can either be an IPv4 or IPv6 address.
  16 //!
  17 //! A TCP connection implements the `Reader` and `Writer` traits, while the TCP
  18 //! listener (socket server) implements the `Listener` and `Acceptor` traits.
  19
  20 #![deny(missing_doc)]
  21
  22 use clone::Clone;
  23 use io::IoResult;
  24 use io::net::ip::SocketAddr;
  25 use io::{Reader, Writer, Listener, Acceptor};
  26 use kinds::Send;
  27 use rt::rtio::{IoFactory, LocalIo, RtioSocket, RtioTcpListener};
  28 use rt::rtio::{RtioTcpAcceptor, RtioTcpStream};
  29
  30 /// A structure which represents a TCP stream between a local socket and a
  31 /// remote socket.
  32 ///
  33 /// # Example
  34 ///
  35 /// ```rust
  36 /// # #[allow(unused_must_use)];
  37 /// use std::io::net::tcp::TcpStream;
  38 /// use std::io::net::ip::{Ipv4Addr, SocketAddr};
  39 ///
  40 /// let addr = SocketAddr { ip: Ipv4Addr(127, 0, 0, 1), port: 34254 };
  41 /// let mut stream = TcpStream::connect(addr);
  42 ///
  43 /// stream.write([1]);
  44 /// let mut buf = [0];
  45 /// stream.read(buf);
  46 /// drop(stream); // close the connection
  47 /// ```
  48 pub struct TcpStream {
  49     priv obj: ~RtioTcpStream:Send
  50 }
  51
  52 impl TcpStream {
  53     fn new(s: ~RtioTcpStream:Send) -> TcpStream {
  54         TcpStream { obj: s }
  55     }
  56
  57     /// Creates a TCP connection to a remote socket address.
  58     ///
  59     /// If no error is encountered, then `Ok(stream)` is returned.
  60     pub fn connect(addr: SocketAddr) -> IoResult<TcpStream> {
  61         LocalIo::maybe_raise(|io| {
  62             io.tcp_connect(addr).map(TcpStream::new)
  63         })
  64     }
  65
  66     /// Returns the socket address of the remote peer of this TCP connection.
  67     pub fn peer_name(&mut self) -> IoResult<SocketAddr> {
  68         self.obj.peer_name()
  69     }
  70
  71     /// Returns the socket address of the local half of this TCP connection.
  72     pub fn socket_name(&mut self) -> IoResult<SocketAddr> {
  73         self.obj.socket_name()
  74     }
  75 }
  76
  77 impl Clone for TcpStream {
  78     /// Creates a new handle to this TCP stream, allowing for simultaneous reads
  79     /// and writes of this connection.
  80     ///
  81     /// The underlying TCP stream will not be closed until all handles to the
  82     /// stream have been deallocated. All handles will also follow the same
  83     /// stream, but two concurrent reads will not receive the same data.
  84     /// Instead, the first read will receive the first packet received, and the
  85     /// second read will receive the second packet.
  86     fn clone(&self) -> TcpStream {
  87         TcpStream { obj: self.obj.clone() }
  88     }
  89 }
  90
  91 impl Reader for TcpStream {
  92     fn read(&mut self, buf: &mut [u8]) -> IoResult<uint> { self.obj.read(buf) }
  93 }
  94
  95 impl Writer for TcpStream {
  96     fn write(&mut self, buf: &[u8]) -> IoResult<()> { self.obj.write(buf) }
  97 }
  98
  99 /// A structure representing a socket server. This listener is used to create a
 100 /// `TcpAcceptor` which can be used to accept sockets on a local port.
 101 ///
 102 /// # Example
 103 ///
 104 /// ```rust
 105 /// # fn main() {}
 106 /// # fn foo() {
 107 /// # #[allow(unused_must_use, dead_code)];
 108 /// use std::io::net::tcp::TcpListener;
 109 /// use std::io::net::ip::{Ipv4Addr, SocketAddr};
 110 /// use std::io::{Acceptor, Listener};
 111 ///
 112 /// let addr = SocketAddr { ip: Ipv4Addr(127, 0, 0, 1), port: 80 };
 113 /// let listener = TcpListener::bind(addr);
 114 ///
 115 /// // bind the listener to the specified address
 116 /// let mut acceptor = listener.listen();
 117 ///
 118 /// // accept connections and process them
 119 /// # fn handle_client<T>(_: T) {}
 120 /// for stream in acceptor.incoming() {
 121 ///     spawn(proc() {
 122 ///         handle_client(stream);
 123 ///     });
 124 /// }
 125 ///
 126 /// // close the socket server
 127 /// drop(acceptor);
 128 /// # }
 129 /// ```
 130 pub struct TcpListener {
 131     priv obj: ~RtioTcpListener:Send
 132 }
 133
 134 impl TcpListener {
 135     /// Creates a new `TcpListener` which will be bound to the specified local
 136     /// socket address. This listener is not ready for accepting connections,
 137     /// `listen` must be called on it before that's possible.
 138     ///
 139     /// Binding with a port number of 0 will request that the OS assigns a port
 140     /// to this listener. The port allocated can be queried via the
 141     /// `socket_name` function.
 142     pub fn bind(addr: SocketAddr) -> IoResult<TcpListener> {
 143         LocalIo::maybe_raise(|io| {
 144             io.tcp_bind(addr).map(|l| TcpListener { obj: l })
 145         })
 146     }
 147
 148     /// Returns the local socket address of this listener.
 149     pub fn socket_name(&mut self) -> IoResult<SocketAddr> {
 150         self.obj.socket_name()
 151     }
 152 }
 153
 154 impl Listener<TcpStream, TcpAcceptor> for TcpListener {
 155     fn listen(self) -> IoResult<TcpAcceptor> {
 156         self.obj.listen().map(|acceptor| TcpAcceptor { obj: acceptor })
 157     }
 158 }
 159
 160 /// The accepting half of a TCP socket server. This structure is created through
 161 /// a `TcpListener`'s `listen` method, and this object can be used to accept new
 162 /// `TcpStream` instances.
 163 pub struct TcpAcceptor {
 164     priv obj: ~RtioTcpAcceptor:Send
 165 }
 166
 167 impl Acceptor<TcpStream> for TcpAcceptor {
 168     fn accept(&mut self) -> IoResult<TcpStream> {
 169         self.obj.accept().map(TcpStream::new)
 170     }
 171 }
 172
 173 #[cfg(test)]
 174 mod test {
 175     use super::*;
 176     use io::net::ip::SocketAddr;
 177     use io::*;
 178     use prelude::*;
 179
 180     // FIXME #11530 this fails on android because tests are run as root
 181     iotest!(fn bind_error() {
 182         let addr = SocketAddr { ip: Ipv4Addr(0, 0, 0, 0), port: 1 };
 183         match TcpListener::bind(addr) {
 184             Ok(..) => fail!(),
 185             Err(e) => assert_eq!(e.kind, PermissionDenied),
 186         }
 187     } #[ignore(cfg(windows))] #[ignore(cfg(target_os = "android"))])
 188
 189     iotest!(fn connect_error() {
 190         let addr = SocketAddr { ip: Ipv4Addr(0, 0, 0, 0), port: 1 };
 191         match TcpStream::connect(addr) {
 192             Ok(..) => fail!(),
 193             Err(e) => assert_eq!(e.kind, ConnectionRefused),
 194         }
 195     })
 196
 197     iotest!(fn smoke_test_ip4() {
 198         let addr = next_test_ip4();
 199         let mut acceptor = TcpListener::bind(addr).listen();
 200
 201         spawn(proc() {
 202             let mut stream = TcpStream::connect(addr);
 203             stream.write([99]).unwrap();
 204         });
 205
 206         let mut stream = acceptor.accept();
 207         let mut buf = [0];
 208         stream.read(buf).unwrap();
 209         assert!(buf[0] == 99);
 210     })
 211
 212     iotest!(fn smoke_test_ip6() {
 213         let addr = next_test_ip6();
 214         let mut acceptor = TcpListener::bind(addr).listen();
 215
 216         spawn(proc() {
 217             let mut stream = TcpStream::connect(addr);
 218             stream.write([99]).unwrap();
 219         });
 220
 221         let mut stream = acceptor.accept();
 222         let mut buf = [0];
 223         stream.read(buf).unwrap();
 224         assert!(buf[0] == 99);
 225     })
 226
 227     iotest!(fn read_eof_ip4() {
 228         let addr = next_test_ip4();
 229         let mut acceptor = TcpListener::bind(addr).listen();
 230
 231         spawn(proc() {
 232             let _stream = TcpStream::connect(addr);
 233             // Close
 234         });
 235
 236         let mut stream = acceptor.accept();
 237         let mut buf = [0];
 238         let nread = stream.read(buf);
 239         assert!(nread.is_err());
 240     })
 241
 242     iotest!(fn read_eof_ip6() {
 243         let addr = next_test_ip6();
 244         let mut acceptor = TcpListener::bind(addr).listen();
 245
 246         spawn(proc() {
 247             let _stream = TcpStream::connect(addr);
 248             // Close
 249         });
 250
 251         let mut stream = acceptor.accept();
 252         let mut buf = [0];
 253         let nread = stream.read(buf);
 254         assert!(nread.is_err());
 255     })
 256
 257     iotest!(fn read_eof_twice_ip4() {
 258         let addr = next_test_ip4();
 259         let mut acceptor = TcpListener::bind(addr).listen();
 260
 261         spawn(proc() {
 262             let _stream = TcpStream::connect(addr);
 263             // Close
 264         });
 265
 266         let mut stream = acceptor.accept();
 267         let mut buf = [0];
 268         let nread = stream.read(buf);
 269         assert!(nread.is_err());
 270
 271         match stream.read(buf) {
 272             Ok(..) => fail!(),
 273             Err(ref e) => {
 274                 assert!(e.kind == NotConnected || e.kind == EndOfFile,
 275                         "unknown kind: {:?}", e.kind);
 276             }
 277         }
 278     })
 279
 280     iotest!(fn read_eof_twice_ip6() {
 281         let addr = next_test_ip6();
 282         let mut acceptor = TcpListener::bind(addr).listen();
 283
 284         spawn(proc() {
 285             let _stream = TcpStream::connect(addr);
 286             // Close
 287         });
 288
 289         let mut stream = acceptor.accept();
 290         let mut buf = [0];
 291         let nread = stream.read(buf);
 292         assert!(nread.is_err());
 293
 294         match stream.read(buf) {
 295             Ok(..) => fail!(),
 296             Err(ref e) => {
 297                 assert!(e.kind == NotConnected || e.kind == EndOfFile,
 298                         "unknown kind: {:?}", e.kind);
 299             }
 300         }
 301     })
 302
 303     iotest!(fn write_close_ip4() {
 304         let addr = next_test_ip4();
 305         let mut acceptor = TcpListener::bind(addr).listen();
 306
 307         spawn(proc() {
 308             let _stream = TcpStream::connect(addr);
 309             // Close
 310         });
 311
 312         let mut stream = acceptor.accept();
 313         let buf = [0];
 314         loop {
 315             match stream.write(buf) {
 316                 Ok(..) => {}
 317                 Err(e) => {
 318                     assert!(e.kind == ConnectionReset ||
 319                             e.kind == BrokenPipe ||
 320                             e.kind == ConnectionAborted,
 321                             "unknown error: {:?}", e);
 322                     break;
 323                 }
 324             }
 325         }
 326     })
 327
 328     iotest!(fn write_close_ip6() {
 329         let addr = next_test_ip6();
 330         let mut acceptor = TcpListener::bind(addr).listen();
 331
 332         spawn(proc() {
 333             let _stream = TcpStream::connect(addr);
 334             // Close
 335         });
 336
 337         let mut stream = acceptor.accept();
 338         let buf = [0];
 339         loop {
 340             match stream.write(buf) {
 341                 Ok(..) => {}
 342                 Err(e) => {
 343                     assert!(e.kind == ConnectionReset ||
 344                             e.kind == BrokenPipe ||
 345                             e.kind == ConnectionAborted,
 346                             "unknown error: {:?}", e);
 347                     break;
 348                 }
 349             }
 350         }
 351     })
 352
 353     iotest!(fn multiple_connect_serial_ip4() {
 354         let addr = next_test_ip4();
 355         let max = 10u;
 356         let mut acceptor = TcpListener::bind(addr).listen();
 357
 358         spawn(proc() {
 359             for _ in range(0, max) {
 360                 let mut stream = TcpStream::connect(addr);
 361                 stream.write([99]).unwrap();
 362             }
 363         });
 364
 365         for ref mut stream in acceptor.incoming().take(max) {
 366             let mut buf = [0];
 367             stream.read(buf).unwrap();
 368             assert_eq!(buf[0], 99);
 369         }
 370     })
 371
 372     iotest!(fn multiple_connect_serial_ip6() {
 373         let addr = next_test_ip6();
 374         let max = 10u;
 375         let mut acceptor = TcpListener::bind(addr).listen();
 376
 377         spawn(proc() {
 378             for _ in range(0, max) {
 379                 let mut stream = TcpStream::connect(addr);
 380                 stream.write([99]).unwrap();
 381             }
 382         });
 383
 384         for ref mut stream in acceptor.incoming().take(max) {
 385             let mut buf = [0];
 386             stream.read(buf).unwrap();
 387             assert_eq!(buf[0], 99);
 388         }
 389     })
 390
 391     iotest!(fn multiple_connect_interleaved_greedy_schedule_ip4() {
 392         let addr = next_test_ip4();
 393         static MAX: int = 10;
 394         let acceptor = TcpListener::bind(addr).listen();
 395
 396         spawn(proc() {
 397             let mut acceptor = acceptor;
 398             for (i, stream) in acceptor.incoming().enumerate().take(MAX as uint) {
 399                 // Start another task to handle the connection
 400                 spawn(proc() {
 401                     let mut stream = stream;
 402                     let mut buf = [0];
 403                     stream.read(buf).unwrap();
 404                     assert!(buf[0] == i as u8);
 405                     debug!("read");
 406                 });
 407             }
 408         });
 409
 410         connect(0, addr);
 411
 412         fn connect(i: int, addr: SocketAddr) {
 413             if i == MAX { return }
 414
 415             spawn(proc() {
 416                 debug!("connecting");
 417                 let mut stream = TcpStream::connect(addr);
 418                 // Connect again before writing
 419                 connect(i + 1, addr);
 420                 debug!("writing");
 421                 stream.write([i as u8]).unwrap();
 422             });
 423         }
 424     })
 425
 426     iotest!(fn multiple_connect_interleaved_greedy_schedule_ip6() {
 427         let addr = next_test_ip6();
 428         static MAX: int = 10;
 429         let acceptor = TcpListener::bind(addr).listen();
 430
 431         spawn(proc() {
 432             let mut acceptor = acceptor;
 433             for (i, stream) in acceptor.incoming().enumerate().take(MAX as uint) {
 434                 // Start another task to handle the connection
 435                 spawn(proc() {
 436                     let mut stream = stream;
 437                     let mut buf = [0];
 438                     stream.read(buf).unwrap();
 439                     assert!(buf[0] == i as u8);
 440                     debug!("read");
 441                 });
 442             }
 443         });
 444
 445         connect(0, addr);
 446
 447         fn connect(i: int, addr: SocketAddr) {
 448             if i == MAX { return }
 449
 450             spawn(proc() {
 451                 debug!("connecting");
 452                 let mut stream = TcpStream::connect(addr);
 453                 // Connect again before writing
 454                 connect(i + 1, addr);
 455                 debug!("writing");
 456                 stream.write([i as u8]).unwrap();
 457             });
 458         }
 459     })
 460
 461     iotest!(fn multiple_connect_interleaved_lazy_schedule_ip4() {
 462         static MAX: int = 10;
 463         let addr = next_test_ip4();
 464         let acceptor = TcpListener::bind(addr).listen();
 465
 466         spawn(proc() {
 467             let mut acceptor = acceptor;
 468             for stream in acceptor.incoming().take(MAX as uint) {
 469                 // Start another task to handle the connection
 470                 spawn(proc() {
 471                     let mut stream = stream;
 472                     let mut buf = [0];
 473                     stream.read(buf).unwrap();
 474                     assert!(buf[0] == 99);
 475                     debug!("read");
 476                 });
 477             }
 478         });
 479
 480         connect(0, addr);
 481
 482         fn connect(i: int, addr: SocketAddr) {
 483             if i == MAX { return }
 484
 485             spawn(proc() {
 486                 debug!("connecting");
 487                 let mut stream = TcpStream::connect(addr);
 488                 // Connect again before writing
 489                 connect(i + 1, addr);
 490                 debug!("writing");
 491                 stream.write([99]).unwrap();
 492             });
 493         }
 494     })
 495
 496     iotest!(fn multiple_connect_interleaved_lazy_schedule_ip6() {
 497         static MAX: int = 10;
 498         let addr = next_test_ip6();
 499         let acceptor = TcpListener::bind(addr).listen();
 500
 501         spawn(proc() {
 502             let mut acceptor = acceptor;
 503             for stream in acceptor.incoming().take(MAX as uint) {
 504                 // Start another task to handle the connection
 505                 spawn(proc() {
 506                     let mut stream = stream;
 507                     let mut buf = [0];
 508                     stream.read(buf).unwrap();
 509                     assert!(buf[0] == 99);
 510                     debug!("read");
 511                 });
 512             }
 513         });
 514
 515         connect(0, addr);
 516
 517         fn connect(i: int, addr: SocketAddr) {
 518             if i == MAX { return }
 519
 520             spawn(proc() {
 521                 debug!("connecting");
 522                 let mut stream = TcpStream::connect(addr);
 523                 // Connect again before writing
 524                 connect(i + 1, addr);
 525                 debug!("writing");
 526                 stream.write([99]).unwrap();
 527             });
 528         }
 529     })
 530
 531     pub fn socket_name(addr: SocketAddr) {
 532         let mut listener = TcpListener::bind(addr).unwrap();
 533
 534         // Make sure socket_name gives
 535         // us the socket we binded to.
 536         let so_name = listener.socket_name();
 537         assert!(so_name.is_ok());
 538         assert_eq!(addr, so_name.unwrap());
 539     }
 540
 541     pub fn peer_name(addr: SocketAddr) {
 542         let acceptor = TcpListener::bind(addr).listen();
 543         spawn(proc() {
 544             let mut acceptor = acceptor;
 545             acceptor.accept().unwrap();
 546         });
 547
 548         let stream = TcpStream::connect(addr);
 549
 550         assert!(stream.is_ok());
 551         let mut stream = stream.unwrap();
 552
 553         // Make sure peer_name gives us the
 554         // address/port of the peer we've
 555         // connected to.
 556         let peer_name = stream.peer_name();
 557         assert!(peer_name.is_ok());
 558         assert_eq!(addr, peer_name.unwrap());
 559     }
 560
 561     iotest!(fn socket_and_peer_name_ip4() {
 562         peer_name(next_test_ip4());
 563         socket_name(next_test_ip4());
 564     })
 565
 566     iotest!(fn socket_and_peer_name_ip6() {
 567         // FIXME: peer name is not consistent
 568         //peer_name(next_test_ip6());
 569         socket_name(next_test_ip6());
 570     })
 571
 572     iotest!(fn partial_read() {
 573         let addr = next_test_ip4();
 574         let (tx, rx) = channel();
 575         spawn(proc() {
 576             let mut srv = TcpListener::bind(addr).listen().unwrap();
 577             tx.send(());
 578             let mut cl = srv.accept().unwrap();
 579             cl.write([10]).unwrap();
 580             let mut b = [0];
 581             cl.read(b).unwrap();
 582             tx.send(());
 583         });
 584
 585         rx.recv();
 586         let mut c = TcpStream::connect(addr).unwrap();
 587         let mut b = [0, ..10];
 588         assert_eq!(c.read(b), Ok(1));
 589         c.write([1]).unwrap();
 590         rx.recv();
 591     })
 592
 593     iotest!(fn double_bind() {
 594         let addr = next_test_ip4();
 595         let listener = TcpListener::bind(addr).unwrap().listen();
 596         assert!(listener.is_ok());
 597         match TcpListener::bind(addr).listen() {
 598             Ok(..) => fail!(),
 599             Err(e) => {
 600                 assert!(e.kind == ConnectionRefused || e.kind == OtherIoError);
 601             }
 602         }
 603     })
 604
 605     iotest!(fn fast_rebind() {
 606         let addr = next_test_ip4();
 607         let (tx, rx) = channel();
 608
 609         spawn(proc() {
 610             rx.recv();
 611             let _stream = TcpStream::connect(addr).unwrap();
 612             // Close
 613             rx.recv();
 614         });
 615
 616         {
 617             let mut acceptor = TcpListener::bind(addr).listen();
 618             tx.send(());
 619             {
 620                 let _stream = acceptor.accept().unwrap();
 621                 // Close client
 622                 tx.send(());
 623             }
 624             // Close listener
 625         }
 626         let _listener = TcpListener::bind(addr);
 627     })
 628
 629     iotest!(fn tcp_clone_smoke() {
 630         let addr = next_test_ip4();
 631         let mut acceptor = TcpListener::bind(addr).listen();
 632
 633         spawn(proc() {
 634             let mut s = TcpStream::connect(addr);
 635             let mut buf = [0, 0];
 636             assert_eq!(s.read(buf), Ok(1));
 637             assert_eq!(buf[0], 1);
 638             s.write([2]).unwrap();
 639         });
 640
 641         let mut s1 = acceptor.accept().unwrap();
 642         let s2 = s1.clone();
 643
 644         let (tx1, rx1) = channel();
 645         let (tx2, rx2) = channel();
 646         spawn(proc() {
 647             let mut s2 = s2;
 648             rx1.recv();
 649             s2.write([1]).unwrap();
 650             tx2.send(());
 651         });
 652         tx1.send(());
 653         let mut buf = [0, 0];
 654         assert_eq!(s1.read(buf), Ok(1));
 655         rx2.recv();
 656     })
 657
 658     iotest!(fn tcp_clone_two_read() {
 659         let addr = next_test_ip6();
 660         let mut acceptor = TcpListener::bind(addr).listen();
 661         let (tx1, rx) = channel();
 662         let tx2 = tx1.clone();
 663
 664         spawn(proc() {
 665             let mut s = TcpStream::connect(addr);
 666             s.write([1]).unwrap();
 667             rx.recv();
 668             s.write([2]).unwrap();
 669             rx.recv();
 670         });
 671
 672         let mut s1 = acceptor.accept().unwrap();
 673         let s2 = s1.clone();
 674
 675         let (done, rx) = channel();
 676         spawn(proc() {
 677             let mut s2 = s2;
 678             let mut buf = [0, 0];
 679             s2.read(buf).unwrap();
 680             tx2.send(());
 681             done.send(());
 682         });
 683         let mut buf = [0, 0];
 684         s1.read(buf).unwrap();
 685         tx1.send(());
 686
 687         rx.recv();
 688     })
 689
 690     iotest!(fn tcp_clone_two_write() {
 691         let addr = next_test_ip4();
 692         let mut acceptor = TcpListener::bind(addr).listen();
 693
 694         spawn(proc() {
 695             let mut s = TcpStream::connect(addr);
 696             let mut buf = [0, 1];
 697             s.read(buf).unwrap();
 698             s.read(buf).unwrap();
 699         });
 700
 701         let mut s1 = acceptor.accept().unwrap();
 702         let s2 = s1.clone();
 703
 704         let (done, rx) = channel();
 705         spawn(proc() {
 706             let mut s2 = s2;
 707             s2.write([1]).unwrap();
 708             done.send(());
 709         });
 710         s1.write([2]).unwrap();
 711
 712         rx.recv();
 713     })
 714
 715     iotest!(fn shutdown_smoke() {
 716         use rt::rtio::RtioTcpStream;
 717
 718         let addr = next_test_ip4();
 719         let a = TcpListener::bind(addr).unwrap().listen();
 720         spawn(proc() {
 721             let mut a = a;
 722             let mut c = a.accept().unwrap();
 723             assert_eq!(c.read_to_end(), Ok(~[]));
 724             c.write([1]).unwrap();
 725         });
 726
 727         let mut s = TcpStream::connect(addr).unwrap();
 728         assert!(s.obj.close_write().is_ok());
 729         assert!(s.write([1]).is_err());
 730         assert_eq!(s.read_to_end(), Ok(~[1]));
 731     })
 732 }
 733