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.
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.
16 use net::{ToSocketAddrs, SocketAddr, Shutdown};
17 use sys_common::io::read_to_end_uninitialized;
18 use sys_common::net as net_imp;
19 use sys_common::{AsInner, FromInner, IntoInner};
22 /// A structure which represents a TCP stream between a local socket and a
25 /// The socket will be closed when the value is dropped.
30 /// use std::io::prelude::*;
31 /// use std::net::TcpStream;
34 /// let mut stream = TcpStream::connect("127.0.0.1:34254").unwrap();
36 /// // ignore the Result
37 /// let _ = stream.write(&[1]);
38 /// let _ = stream.read(&mut [0; 128]); // ignore here too
39 /// } // the stream is closed here
41 #[stable(feature = "rust1", since = "1.0.0")]
42 pub struct TcpStream(net_imp::TcpStream);
44 /// A structure representing a socket server.
49 /// use std::net::{TcpListener, TcpStream};
52 /// let listener = TcpListener::bind("127.0.0.1:80").unwrap();
54 /// fn handle_client(stream: TcpStream) {
58 /// // accept connections and process them, spawning a new thread for each one
59 /// for stream in listener.incoming() {
62 /// thread::spawn(move|| {
63 /// // connection succeeded
64 /// handle_client(stream)
67 /// Err(e) => { /* connection failed */ }
71 /// // close the socket server
74 #[stable(feature = "rust1", since = "1.0.0")]
75 pub struct TcpListener(net_imp::TcpListener);
77 /// An infinite iterator over the connections from a `TcpListener`.
79 /// This iterator will infinitely yield `Some` of the accepted connections. It
80 /// is equivalent to calling `accept` in a loop.
81 #[stable(feature = "rust1", since = "1.0.0")]
82 pub struct Incoming<'a> { listener: &'a TcpListener }
85 /// Opens a TCP connection to a remote host.
87 /// `addr` is an address of the remote host. Anything which implements
88 /// `ToSocketAddrs` trait can be supplied for the address; see this trait
89 /// documentation for concrete examples.
90 #[stable(feature = "rust1", since = "1.0.0")]
91 pub fn connect<A: ToSocketAddrs>(addr: A) -> io::Result<TcpStream> {
92 super::each_addr(addr, net_imp::TcpStream::connect).map(TcpStream)
95 /// Returns the socket address of the remote peer of this TCP connection.
96 #[stable(feature = "rust1", since = "1.0.0")]
97 pub fn peer_addr(&self) -> io::Result<SocketAddr> {
101 /// Returns the socket address of the local half of this TCP connection.
102 #[stable(feature = "rust1", since = "1.0.0")]
103 pub fn local_addr(&self) -> io::Result<SocketAddr> {
107 /// Shuts down the read, write, or both halves of this connection.
109 /// This function will cause all pending and future I/O on the specified
110 /// portions to return immediately with an appropriate value (see the
111 /// documentation of `Shutdown`).
112 #[stable(feature = "rust1", since = "1.0.0")]
113 pub fn shutdown(&self, how: Shutdown) -> io::Result<()> {
117 /// Creates a new independently owned handle to the underlying socket.
119 /// The returned `TcpStream` is a reference to the same stream that this
120 /// object references. Both handles will read and write the same stream of
121 /// data, and options set on one stream will be propagated to the other
123 #[stable(feature = "rust1", since = "1.0.0")]
124 pub fn try_clone(&self) -> io::Result<TcpStream> {
125 self.0.duplicate().map(TcpStream)
128 /// Sets the read timeout to the timeout specified.
130 /// If the value specified is `None`, then `read` calls will block
131 /// indefinitely. It is an error to pass the zero `Duration` to this
136 /// Platforms may return a different error code whenever a read times out as
137 /// a result of setting this option. For example Unix typically returns an
138 /// error of the kind `WouldBlock`, but Windows may return `TimedOut`.
139 #[stable(feature = "socket_timeout", since = "1.4.0")]
140 pub fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
141 self.0.set_read_timeout(dur)
144 /// Sets the write timeout to the timeout specified.
146 /// If the value specified is `None`, then `write` calls will block
147 /// indefinitely. It is an error to pass the zero `Duration` to this
152 /// Platforms may return a different error code whenever a write times out
153 /// as a result of setting this option. For example Unix typically returns
154 /// an error of the kind `WouldBlock`, but Windows may return `TimedOut`.
155 #[stable(feature = "socket_timeout", since = "1.4.0")]
156 pub fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
157 self.0.set_write_timeout(dur)
160 /// Returns the read timeout of this socket.
162 /// If the timeout is `None`, then `read` calls will block indefinitely.
166 /// Some platforms do not provide access to the current timeout.
167 #[stable(feature = "socket_timeout", since = "1.4.0")]
168 pub fn read_timeout(&self) -> io::Result<Option<Duration>> {
169 self.0.read_timeout()
172 /// Returns the write timeout of this socket.
174 /// If the timeout is `None`, then `write` calls will block indefinitely.
178 /// Some platforms do not provide access to the current timeout.
179 #[stable(feature = "socket_timeout", since = "1.4.0")]
180 pub fn write_timeout(&self) -> io::Result<Option<Duration>> {
181 self.0.write_timeout()
184 /// Sets the value of the `TCP_NODELAY` option on this socket.
186 /// If set, this option disables the Nagle algorithm. This means that
187 /// segments are always sent as soon as possible, even if there is only a
188 /// small amount of data. When not set, data is buffered until there is a
189 /// sufficient amount to send out, thereby avoiding the frequent sending of
191 #[stable(feature = "net2_mutators", since = "1.9.0")]
192 pub fn set_nodelay(&self, nodelay: bool) -> io::Result<()> {
193 self.0.set_nodelay(nodelay)
196 /// Gets the value of the `TCP_NODELAY` option on this socket.
198 /// For more information about this option, see [`set_nodelay`][link].
200 /// [link]: #tymethod.set_nodelay
201 #[stable(feature = "net2_mutators", since = "1.9.0")]
202 pub fn nodelay(&self) -> io::Result<bool> {
206 /// Sets the value for the `IP_TTL` option on this socket.
208 /// This value sets the time-to-live field that is used in every packet sent
209 /// from this socket.
210 #[stable(feature = "net2_mutators", since = "1.9.0")]
211 pub fn set_ttl(&self, ttl: u32) -> io::Result<()> {
215 /// Gets the value of the `IP_TTL` option for this socket.
217 /// For more information about this option, see [`set_ttl`][link].
219 /// [link]: #tymethod.set_ttl
220 #[stable(feature = "net2_mutators", since = "1.9.0")]
221 pub fn ttl(&self) -> io::Result<u32> {
225 /// Sets the value for the `IPV6_V6ONLY` option on this socket.
227 /// If this is set to `true` then the socket is restricted to sending and
228 /// receiving IPv6 packets only. If this is the case, an IPv4 and an IPv6
229 /// application can each bind the same port at the same time.
231 /// If this is set to `false` then the socket can be used to send and
232 /// receive packets from an IPv4-mapped IPv6 address.
233 #[stable(feature = "net2_mutators", since = "1.9.0")]
234 pub fn set_only_v6(&self, only_v6: bool) -> io::Result<()> {
235 self.0.set_only_v6(only_v6)
238 /// Gets the value of the `IPV6_V6ONLY` option for this socket.
240 /// For more information about this option, see [`set_only_v6`][link].
242 /// [link]: #tymethod.set_only_v6
243 #[stable(feature = "net2_mutators", since = "1.9.0")]
244 pub fn only_v6(&self) -> io::Result<bool> {
248 /// Get the value of the `SO_ERROR` option on this socket.
250 /// This will retrieve the stored error in the underlying socket, clearing
251 /// the field in the process. This can be useful for checking errors between
253 #[stable(feature = "net2_mutators", since = "1.9.0")]
254 pub fn take_error(&self) -> io::Result<Option<io::Error>> {
258 /// Moves this TCP stream into or out of nonblocking mode.
260 /// On Unix this corresponds to calling fcntl, and on Windows this
261 /// corresponds to calling ioctlsocket.
262 #[stable(feature = "net2_mutators", since = "1.9.0")]
263 pub fn set_nonblocking(&self, nonblocking: bool) -> io::Result<()> {
264 self.0.set_nonblocking(nonblocking)
268 #[stable(feature = "rust1", since = "1.0.0")]
269 impl Read for TcpStream {
270 fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> { self.0.read(buf) }
271 fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
272 unsafe { read_to_end_uninitialized(self, buf) }
275 #[stable(feature = "rust1", since = "1.0.0")]
276 impl Write for TcpStream {
277 fn write(&mut self, buf: &[u8]) -> io::Result<usize> { self.0.write(buf) }
278 fn flush(&mut self) -> io::Result<()> { Ok(()) }
280 #[stable(feature = "rust1", since = "1.0.0")]
281 impl<'a> Read for &'a TcpStream {
282 fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> { self.0.read(buf) }
283 fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
284 unsafe { read_to_end_uninitialized(self, buf) }
287 #[stable(feature = "rust1", since = "1.0.0")]
288 impl<'a> Write for &'a TcpStream {
289 fn write(&mut self, buf: &[u8]) -> io::Result<usize> { self.0.write(buf) }
290 fn flush(&mut self) -> io::Result<()> { Ok(()) }
293 impl AsInner<net_imp::TcpStream> for TcpStream {
294 fn as_inner(&self) -> &net_imp::TcpStream { &self.0 }
297 impl FromInner<net_imp::TcpStream> for TcpStream {
298 fn from_inner(inner: net_imp::TcpStream) -> TcpStream { TcpStream(inner) }
301 impl IntoInner<net_imp::TcpStream> for TcpStream {
302 fn into_inner(self) -> net_imp::TcpStream { self.0 }
305 #[stable(feature = "rust1", since = "1.0.0")]
306 impl fmt::Debug for TcpStream {
307 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
313 /// Creates a new `TcpListener` which will be bound to the specified
316 /// The returned listener is ready for accepting connections.
318 /// Binding with a port number of 0 will request that the OS assigns a port
319 /// to this listener. The port allocated can be queried via the
320 /// `local_addr` method.
322 /// The address type can be any implementor of `ToSocketAddrs` trait. See
323 /// its documentation for concrete examples.
324 #[stable(feature = "rust1", since = "1.0.0")]
325 pub fn bind<A: ToSocketAddrs>(addr: A) -> io::Result<TcpListener> {
326 super::each_addr(addr, net_imp::TcpListener::bind).map(TcpListener)
329 /// Returns the local socket address of this listener.
330 #[stable(feature = "rust1", since = "1.0.0")]
331 pub fn local_addr(&self) -> io::Result<SocketAddr> {
335 /// Creates a new independently owned handle to the underlying socket.
337 /// The returned `TcpListener` is a reference to the same socket that this
338 /// object references. Both handles can be used to accept incoming
339 /// connections and options set on one listener will affect the other.
340 #[stable(feature = "rust1", since = "1.0.0")]
341 pub fn try_clone(&self) -> io::Result<TcpListener> {
342 self.0.duplicate().map(TcpListener)
345 /// Accept a new incoming connection from this listener.
347 /// This function will block the calling thread until a new TCP connection
348 /// is established. When established, the corresponding `TcpStream` and the
349 /// remote peer's address will be returned.
350 #[stable(feature = "rust1", since = "1.0.0")]
351 pub fn accept(&self) -> io::Result<(TcpStream, SocketAddr)> {
352 self.0.accept().map(|(a, b)| (TcpStream(a), b))
355 /// Returns an iterator over the connections being received on this
358 /// The returned iterator will never return `None` and will also not yield
359 /// the peer's `SocketAddr` structure.
360 #[stable(feature = "rust1", since = "1.0.0")]
361 pub fn incoming(&self) -> Incoming {
362 Incoming { listener: self }
365 /// Sets the value for the `IP_TTL` option on this socket.
367 /// This value sets the time-to-live field that is used in every packet sent
368 /// from this socket.
369 #[stable(feature = "net2_mutators", since = "1.9.0")]
370 pub fn set_ttl(&self, ttl: u32) -> io::Result<()> {
374 /// Gets the value of the `IP_TTL` option for this socket.
376 /// For more information about this option, see [`set_ttl`][link].
378 /// [link]: #tymethod.set_ttl
379 #[stable(feature = "net2_mutators", since = "1.9.0")]
380 pub fn ttl(&self) -> io::Result<u32> {
384 /// Sets the value for the `IPV6_V6ONLY` option on this socket.
386 /// If this is set to `true` then the socket is restricted to sending and
387 /// receiving IPv6 packets only. In this case two IPv4 and IPv6 applications
388 /// can bind the same port at the same time.
390 /// If this is set to `false` then the socket can be used to send and
391 /// receive packets from an IPv4-mapped IPv6 address.
392 #[stable(feature = "net2_mutators", since = "1.9.0")]
393 pub fn set_only_v6(&self, only_v6: bool) -> io::Result<()> {
394 self.0.set_only_v6(only_v6)
397 /// Gets the value of the `IPV6_V6ONLY` option for this socket.
399 /// For more information about this option, see [`set_only_v6`][link].
401 /// [link]: #tymethod.set_only_v6
402 #[stable(feature = "net2_mutators", since = "1.9.0")]
403 pub fn only_v6(&self) -> io::Result<bool> {
407 /// Get the value of the `SO_ERROR` option on this socket.
409 /// This will retrieve the stored error in the underlying socket, clearing
410 /// the field in the process. This can be useful for checking errors between
412 #[stable(feature = "net2_mutators", since = "1.9.0")]
413 pub fn take_error(&self) -> io::Result<Option<io::Error>> {
417 /// Moves this TCP stream into or out of nonblocking mode.
419 /// On Unix this corresponds to calling fcntl, and on Windows this
420 /// corresponds to calling ioctlsocket.
421 #[stable(feature = "net2_mutators", since = "1.9.0")]
422 pub fn set_nonblocking(&self, nonblocking: bool) -> io::Result<()> {
423 self.0.set_nonblocking(nonblocking)
427 #[stable(feature = "rust1", since = "1.0.0")]
428 impl<'a> Iterator for Incoming<'a> {
429 type Item = io::Result<TcpStream>;
430 fn next(&mut self) -> Option<io::Result<TcpStream>> {
431 Some(self.listener.accept().map(|p| p.0))
435 impl AsInner<net_imp::TcpListener> for TcpListener {
436 fn as_inner(&self) -> &net_imp::TcpListener { &self.0 }
439 impl FromInner<net_imp::TcpListener> for TcpListener {
440 fn from_inner(inner: net_imp::TcpListener) -> TcpListener {
445 impl IntoInner<net_imp::TcpListener> for TcpListener {
446 fn into_inner(self) -> net_imp::TcpListener { self.0 }
449 #[stable(feature = "rust1", since = "1.0.0")]
450 impl fmt::Debug for TcpListener {
451 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
463 use net::test::{next_test_ip4, next_test_ip6};
464 use sync::mpsc::channel;
465 use sys_common::AsInner;
466 use time::{Instant, Duration};
469 fn each_ip(f: &mut FnMut(SocketAddr)) {
478 Err(e) => panic!("received error for `{}`: {}", stringify!($e), e),
485 match TcpListener::bind("1.1.1.1:9999") {
488 assert_eq!(e.kind(), ErrorKind::AddrNotAvailable),
494 match TcpStream::connect("0.0.0.0:1") {
496 Err(e) => assert!(e.kind() == ErrorKind::ConnectionRefused ||
497 e.kind() == ErrorKind::InvalidInput ||
498 e.kind() == ErrorKind::AddrInUse ||
499 e.kind() == ErrorKind::AddrNotAvailable,
500 "bad error: {} {:?}", e, e.kind()),
505 fn listen_localhost() {
506 let socket_addr = next_test_ip4();
507 let listener = t!(TcpListener::bind(&socket_addr));
509 let _t = thread::spawn(move || {
510 let mut stream = t!(TcpStream::connect(&("localhost",
511 socket_addr.port())));
512 t!(stream.write(&[144]));
515 let mut stream = t!(listener.accept()).0;
517 t!(stream.read(&mut buf));
518 assert!(buf[0] == 144);
522 fn connect_loopback() {
523 each_ip(&mut |addr| {
524 let acceptor = t!(TcpListener::bind(&addr));
526 let _t = thread::spawn(move|| {
527 let host = match addr {
528 SocketAddr::V4(..) => "127.0.0.1",
529 SocketAddr::V6(..) => "::1",
531 let mut stream = t!(TcpStream::connect(&(host, addr.port())));
532 t!(stream.write(&[66]));
535 let mut stream = t!(acceptor.accept()).0;
537 t!(stream.read(&mut buf));
538 assert!(buf[0] == 66);
544 each_ip(&mut |addr| {
545 let acceptor = t!(TcpListener::bind(&addr));
547 let (tx, rx) = channel();
548 let _t = thread::spawn(move|| {
549 let mut stream = t!(TcpStream::connect(&addr));
550 t!(stream.write(&[99]));
551 tx.send(t!(stream.local_addr())).unwrap();
554 let (mut stream, addr) = t!(acceptor.accept());
556 t!(stream.read(&mut buf));
557 assert!(buf[0] == 99);
558 assert_eq!(addr, t!(rx.recv()));
564 each_ip(&mut |addr| {
565 let acceptor = t!(TcpListener::bind(&addr));
567 let _t = thread::spawn(move|| {
568 let _stream = t!(TcpStream::connect(&addr));
572 let mut stream = t!(acceptor.accept()).0;
574 let nread = t!(stream.read(&mut buf));
575 assert_eq!(nread, 0);
576 let nread = t!(stream.read(&mut buf));
577 assert_eq!(nread, 0);
583 each_ip(&mut |addr| {
584 let acceptor = t!(TcpListener::bind(&addr));
586 let (tx, rx) = channel();
587 let _t = thread::spawn(move|| {
588 drop(t!(TcpStream::connect(&addr)));
589 tx.send(()).unwrap();
592 let mut stream = t!(acceptor.accept()).0;
595 match stream.write(&buf) {
598 assert!(e.kind() == ErrorKind::ConnectionReset ||
599 e.kind() == ErrorKind::BrokenPipe ||
600 e.kind() == ErrorKind::ConnectionAborted,
601 "unknown error: {}", e);
608 fn multiple_connect_serial() {
609 each_ip(&mut |addr| {
611 let acceptor = t!(TcpListener::bind(&addr));
613 let _t = thread::spawn(move|| {
615 let mut stream = t!(TcpStream::connect(&addr));
616 t!(stream.write(&[99]));
620 for stream in acceptor.incoming().take(max) {
621 let mut stream = t!(stream);
623 t!(stream.read(&mut buf));
624 assert_eq!(buf[0], 99);
630 fn multiple_connect_interleaved_greedy_schedule() {
631 const MAX: usize = 10;
632 each_ip(&mut |addr| {
633 let acceptor = t!(TcpListener::bind(&addr));
635 let _t = thread::spawn(move|| {
636 let acceptor = acceptor;
637 for (i, stream) in acceptor.incoming().enumerate().take(MAX) {
638 // Start another thread to handle the connection
639 let _t = thread::spawn(move|| {
640 let mut stream = t!(stream);
642 t!(stream.read(&mut buf));
643 assert!(buf[0] == i as u8);
651 fn connect(i: usize, addr: SocketAddr) {
652 if i == MAX { return }
654 let t = thread::spawn(move|| {
655 let mut stream = t!(TcpStream::connect(&addr));
656 // Connect again before writing
657 connect(i + 1, addr);
658 t!(stream.write(&[i as u8]));
660 t.join().ok().unwrap();
665 fn multiple_connect_interleaved_lazy_schedule() {
666 const MAX: usize = 10;
667 each_ip(&mut |addr| {
668 let acceptor = t!(TcpListener::bind(&addr));
670 let _t = thread::spawn(move|| {
671 for stream in acceptor.incoming().take(MAX) {
672 // Start another thread to handle the connection
673 let _t = thread::spawn(move|| {
674 let mut stream = t!(stream);
676 t!(stream.read(&mut buf));
677 assert!(buf[0] == 99);
685 fn connect(i: usize, addr: SocketAddr) {
686 if i == MAX { return }
688 let t = thread::spawn(move|| {
689 let mut stream = t!(TcpStream::connect(&addr));
690 connect(i + 1, addr);
691 t!(stream.write(&[99]));
693 t.join().ok().unwrap();
698 fn socket_and_peer_name() {
699 each_ip(&mut |addr| {
700 let listener = t!(TcpListener::bind(&addr));
701 let so_name = t!(listener.local_addr());
702 assert_eq!(addr, so_name);
703 let _t = thread::spawn(move|| {
704 t!(listener.accept());
707 let stream = t!(TcpStream::connect(&addr));
708 assert_eq!(addr, t!(stream.peer_addr()));
714 each_ip(&mut |addr| {
715 let (tx, rx) = channel();
716 let srv = t!(TcpListener::bind(&addr));
717 let _t = thread::spawn(move|| {
718 let mut cl = t!(srv.accept()).0;
719 cl.write(&[10]).unwrap();
722 tx.send(()).unwrap();
725 let mut c = t!(TcpStream::connect(&addr));
727 assert_eq!(c.read(&mut b).unwrap(), 1);
735 each_ip(&mut |addr| {
736 let _listener = t!(TcpListener::bind(&addr));
737 match TcpListener::bind(&addr) {
740 assert!(e.kind() == ErrorKind::ConnectionRefused ||
741 e.kind() == ErrorKind::Other ||
742 e.kind() == ErrorKind::AddrInUse,
743 "unknown error: {} {:?}", e, e.kind());
751 each_ip(&mut |addr| {
752 let acceptor = t!(TcpListener::bind(&addr));
754 let _t = thread::spawn(move|| {
755 t!(TcpStream::connect(&addr));
758 t!(acceptor.accept());
760 t!(TcpListener::bind(&addr));
765 fn tcp_clone_smoke() {
766 each_ip(&mut |addr| {
767 let acceptor = t!(TcpListener::bind(&addr));
769 let _t = thread::spawn(move|| {
770 let mut s = t!(TcpStream::connect(&addr));
771 let mut buf = [0, 0];
772 assert_eq!(s.read(&mut buf).unwrap(), 1);
773 assert_eq!(buf[0], 1);
777 let mut s1 = t!(acceptor.accept()).0;
778 let s2 = t!(s1.try_clone());
780 let (tx1, rx1) = channel();
781 let (tx2, rx2) = channel();
782 let _t = thread::spawn(move|| {
786 tx2.send(()).unwrap();
788 tx1.send(()).unwrap();
789 let mut buf = [0, 0];
790 assert_eq!(s1.read(&mut buf).unwrap(), 1);
796 fn tcp_clone_two_read() {
797 each_ip(&mut |addr| {
798 let acceptor = t!(TcpListener::bind(&addr));
799 let (tx1, rx) = channel();
800 let tx2 = tx1.clone();
802 let _t = thread::spawn(move|| {
803 let mut s = t!(TcpStream::connect(&addr));
810 let mut s1 = t!(acceptor.accept()).0;
811 let s2 = t!(s1.try_clone());
813 let (done, rx) = channel();
814 let _t = thread::spawn(move|| {
816 let mut buf = [0, 0];
817 t!(s2.read(&mut buf));
818 tx2.send(()).unwrap();
819 done.send(()).unwrap();
821 let mut buf = [0, 0];
822 t!(s1.read(&mut buf));
823 tx1.send(()).unwrap();
830 fn tcp_clone_two_write() {
831 each_ip(&mut |addr| {
832 let acceptor = t!(TcpListener::bind(&addr));
834 let _t = thread::spawn(move|| {
835 let mut s = t!(TcpStream::connect(&addr));
836 let mut buf = [0, 1];
837 t!(s.read(&mut buf));
838 t!(s.read(&mut buf));
841 let mut s1 = t!(acceptor.accept()).0;
842 let s2 = t!(s1.try_clone());
844 let (done, rx) = channel();
845 let _t = thread::spawn(move|| {
848 done.send(()).unwrap();
857 fn shutdown_smoke() {
858 each_ip(&mut |addr| {
859 let a = t!(TcpListener::bind(&addr));
860 let _t = thread::spawn(move|| {
861 let mut c = t!(a.accept()).0;
863 assert_eq!(c.read(&mut b).unwrap(), 0);
867 let mut s = t!(TcpStream::connect(&addr));
868 t!(s.shutdown(Shutdown::Write));
869 assert!(s.write(&[1]).is_err());
871 assert_eq!(t!(s.read(&mut b)), 1);
877 fn close_readwrite_smoke() {
878 each_ip(&mut |addr| {
879 let a = t!(TcpListener::bind(&addr));
880 let (tx, rx) = channel::<()>();
881 let _t = thread::spawn(move|| {
882 let _s = t!(a.accept());
887 let mut s = t!(TcpStream::connect(&addr));
888 let mut s2 = t!(s.try_clone());
890 // closing should prevent reads/writes
891 t!(s.shutdown(Shutdown::Write));
892 assert!(s.write(&[0]).is_err());
893 t!(s.shutdown(Shutdown::Read));
894 assert_eq!(s.read(&mut b).unwrap(), 0);
896 // closing should affect previous handles
897 assert!(s2.write(&[0]).is_err());
898 assert_eq!(s2.read(&mut b).unwrap(), 0);
900 // closing should affect new handles
901 let mut s3 = t!(s.try_clone());
902 assert!(s3.write(&[0]).is_err());
903 assert_eq!(s3.read(&mut b).unwrap(), 0);
905 // make sure these don't die
906 let _ = s2.shutdown(Shutdown::Read);
907 let _ = s2.shutdown(Shutdown::Write);
908 let _ = s3.shutdown(Shutdown::Read);
909 let _ = s3.shutdown(Shutdown::Write);
915 fn close_read_wakes_up() {
916 each_ip(&mut |addr| {
917 let a = t!(TcpListener::bind(&addr));
918 let (tx1, rx) = channel::<()>();
919 let _t = thread::spawn(move|| {
920 let _s = t!(a.accept());
924 let s = t!(TcpStream::connect(&addr));
925 let s2 = t!(s.try_clone());
926 let (tx, rx) = channel();
927 let _t = thread::spawn(move|| {
929 assert_eq!(t!(s2.read(&mut [0])), 0);
930 tx.send(()).unwrap();
932 // this should wake up the child thread
933 t!(s.shutdown(Shutdown::Read));
935 // this test will never finish if the child doesn't wake up
942 fn clone_while_reading() {
943 each_ip(&mut |addr| {
944 let accept = t!(TcpListener::bind(&addr));
946 // Enqueue a thread to write to a socket
947 let (tx, rx) = channel();
948 let (txdone, rxdone) = channel();
949 let txdone2 = txdone.clone();
950 let _t = thread::spawn(move|| {
951 let mut tcp = t!(TcpStream::connect(&addr));
954 txdone2.send(()).unwrap();
957 // Spawn off a reading clone
958 let tcp = t!(accept.accept()).0;
959 let tcp2 = t!(tcp.try_clone());
960 let txdone3 = txdone.clone();
961 let _t = thread::spawn(move|| {
963 t!(tcp2.read(&mut [0]));
964 txdone3.send(()).unwrap();
967 // Try to ensure that the reading clone is indeed reading
972 // clone the handle again while it's reading, then let it finish the
974 let _ = t!(tcp.try_clone());
975 tx.send(()).unwrap();
976 rxdone.recv().unwrap();
977 rxdone.recv().unwrap();
982 fn clone_accept_smoke() {
983 each_ip(&mut |addr| {
984 let a = t!(TcpListener::bind(&addr));
985 let a2 = t!(a.try_clone());
987 let _t = thread::spawn(move|| {
988 let _ = TcpStream::connect(&addr);
990 let _t = thread::spawn(move|| {
991 let _ = TcpStream::connect(&addr);
1000 fn clone_accept_concurrent() {
1001 each_ip(&mut |addr| {
1002 let a = t!(TcpListener::bind(&addr));
1003 let a2 = t!(a.try_clone());
1005 let (tx, rx) = channel();
1006 let tx2 = tx.clone();
1008 let _t = thread::spawn(move|| {
1009 tx.send(t!(a.accept())).unwrap();
1011 let _t = thread::spawn(move|| {
1012 tx2.send(t!(a2.accept())).unwrap();
1015 let _t = thread::spawn(move|| {
1016 let _ = TcpStream::connect(&addr);
1018 let _t = thread::spawn(move|| {
1019 let _ = TcpStream::connect(&addr);
1029 let name = if cfg!(windows) {"socket"} else {"fd"};
1030 let socket_addr = next_test_ip4();
1032 let listener = t!(TcpListener::bind(&socket_addr));
1033 let listener_inner = listener.0.socket().as_inner();
1034 let compare = format!("TcpListener {{ addr: {:?}, {}: {:?} }}",
1035 socket_addr, name, listener_inner);
1036 assert_eq!(format!("{:?}", listener), compare);
1038 let stream = t!(TcpStream::connect(&("localhost",
1039 socket_addr.port())));
1040 let stream_inner = stream.0.socket().as_inner();
1041 let compare = format!("TcpStream {{ addr: {:?}, \
1042 peer: {:?}, {}: {:?} }}",
1043 stream.local_addr().unwrap(),
1044 stream.peer_addr().unwrap(),
1047 assert_eq!(format!("{:?}", stream), compare);
1050 // FIXME: re-enabled bitrig/openbsd tests once their socket timeout code
1051 // no longer has rounding errors.
1052 #[cfg_attr(any(target_os = "bitrig", target_os = "netbsd", target_os = "openbsd"), ignore)]
1055 let addr = next_test_ip4();
1056 let listener = t!(TcpListener::bind(&addr));
1058 let stream = t!(TcpStream::connect(&("localhost", addr.port())));
1059 let dur = Duration::new(15410, 0);
1061 assert_eq!(None, t!(stream.read_timeout()));
1063 t!(stream.set_read_timeout(Some(dur)));
1064 assert_eq!(Some(dur), t!(stream.read_timeout()));
1066 assert_eq!(None, t!(stream.write_timeout()));
1068 t!(stream.set_write_timeout(Some(dur)));
1069 assert_eq!(Some(dur), t!(stream.write_timeout()));
1071 t!(stream.set_read_timeout(None));
1072 assert_eq!(None, t!(stream.read_timeout()));
1074 t!(stream.set_write_timeout(None));
1075 assert_eq!(None, t!(stream.write_timeout()));
1080 fn test_read_timeout() {
1081 let addr = next_test_ip4();
1082 let listener = t!(TcpListener::bind(&addr));
1084 let mut stream = t!(TcpStream::connect(&("localhost", addr.port())));
1085 t!(stream.set_read_timeout(Some(Duration::from_millis(1000))));
1087 let mut buf = [0; 10];
1088 let start = Instant::now();
1089 let kind = stream.read(&mut buf).err().expect("expected error").kind();
1090 assert!(kind == ErrorKind::WouldBlock || kind == ErrorKind::TimedOut);
1091 assert!(start.elapsed() > Duration::from_millis(400));
1096 fn test_read_with_timeout() {
1097 let addr = next_test_ip4();
1098 let listener = t!(TcpListener::bind(&addr));
1100 let mut stream = t!(TcpStream::connect(&("localhost", addr.port())));
1101 t!(stream.set_read_timeout(Some(Duration::from_millis(1000))));
1103 let mut other_end = t!(listener.accept()).0;
1104 t!(other_end.write_all(b"hello world"));
1106 let mut buf = [0; 11];
1107 t!(stream.read(&mut buf));
1108 assert_eq!(b"hello world", &buf[..]);
1110 let start = Instant::now();
1111 let kind = stream.read(&mut buf).err().expect("expected error").kind();
1112 assert!(kind == ErrorKind::WouldBlock || kind == ErrorKind::TimedOut);
1113 assert!(start.elapsed() > Duration::from_millis(400));
1119 let addr = next_test_ip4();
1120 let _listener = t!(TcpListener::bind(&addr));
1122 let stream = t!(TcpStream::connect(&("localhost", addr.port())));
1124 assert_eq!(false, t!(stream.nodelay()));
1125 t!(stream.set_nodelay(true));
1126 assert_eq!(true, t!(stream.nodelay()));
1127 t!(stream.set_nodelay(false));
1128 assert_eq!(false, t!(stream.nodelay()));
1135 let addr = next_test_ip4();
1136 let listener = t!(TcpListener::bind(&addr));
1138 t!(listener.set_ttl(ttl));
1139 assert_eq!(ttl, t!(listener.ttl()));
1141 let stream = t!(TcpStream::connect(&("localhost", addr.port())));
1143 t!(stream.set_ttl(ttl));
1144 assert_eq!(ttl, t!(stream.ttl()));
1148 fn set_nonblocking() {
1149 let addr = next_test_ip4();
1150 let listener = t!(TcpListener::bind(&addr));
1152 t!(listener.set_nonblocking(true));
1153 t!(listener.set_nonblocking(false));
1155 let mut stream = t!(TcpStream::connect(&("localhost", addr.port())));
1157 t!(stream.set_nonblocking(false));
1158 t!(stream.set_nonblocking(true));
1161 match stream.read(&mut buf) {
1162 Ok(_) => panic!("expected error"),
1163 Err(ref e) if e.kind() == ErrorKind::WouldBlock => {}
1164 Err(e) => panic!("unexpected error {}", e),