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::net as net_imp;
18 use sys_common::{AsInner, FromInner, IntoInner};
21 /// A structure which represents a TCP stream between a local socket and a
24 /// The socket will be closed when the value is dropped.
29 /// use std::io::prelude::*;
30 /// use std::net::TcpStream;
33 /// let mut stream = TcpStream::connect("127.0.0.1:34254").unwrap();
35 /// // ignore the Result
36 /// let _ = stream.write(&[1]);
37 /// let _ = stream.read(&mut [0; 128]); // ignore here too
38 /// } // the stream is closed here
40 #[stable(feature = "rust1", since = "1.0.0")]
41 pub struct TcpStream(net_imp::TcpStream);
43 /// A structure representing a socket server.
48 /// use std::net::{TcpListener, TcpStream};
51 /// let listener = TcpListener::bind("127.0.0.1:80").unwrap();
53 /// fn handle_client(stream: TcpStream) {
57 /// // accept connections and process them, spawning a new thread for each one
58 /// for stream in listener.incoming() {
61 /// thread::spawn(move|| {
62 /// // connection succeeded
63 /// handle_client(stream)
66 /// Err(e) => { /* connection failed */ }
70 /// // close the socket server
73 #[stable(feature = "rust1", since = "1.0.0")]
74 pub struct TcpListener(net_imp::TcpListener);
76 /// An infinite iterator over the connections from a `TcpListener`.
78 /// This iterator will infinitely yield `Some` of the accepted connections. It
79 /// is equivalent to calling `accept` in a loop.
80 #[stable(feature = "rust1", since = "1.0.0")]
81 pub struct Incoming<'a> { listener: &'a TcpListener }
84 /// Opens a TCP connection to a remote host.
86 /// `addr` is an address of the remote host. Anything which implements
87 /// `ToSocketAddrs` trait can be supplied for the address; see this trait
88 /// documentation for concrete examples.
89 #[stable(feature = "rust1", since = "1.0.0")]
90 pub fn connect<A: ToSocketAddrs>(addr: A) -> io::Result<TcpStream> {
91 super::each_addr(addr, net_imp::TcpStream::connect).map(TcpStream)
94 /// Returns the socket address of the remote peer of this TCP connection.
95 #[stable(feature = "rust1", since = "1.0.0")]
96 pub fn peer_addr(&self) -> io::Result<SocketAddr> {
100 /// Returns the socket address of the local half of this TCP connection.
101 #[stable(feature = "rust1", since = "1.0.0")]
102 pub fn local_addr(&self) -> io::Result<SocketAddr> {
106 /// Shuts down the read, write, or both halves of this connection.
108 /// This function will cause all pending and future I/O on the specified
109 /// portions to return immediately with an appropriate value (see the
110 /// documentation of `Shutdown`).
111 #[stable(feature = "rust1", since = "1.0.0")]
112 pub fn shutdown(&self, how: Shutdown) -> io::Result<()> {
116 /// Creates a new independently owned handle to the underlying socket.
118 /// The returned `TcpStream` is a reference to the same stream that this
119 /// object references. Both handles will read and write the same stream of
120 /// data, and options set on one stream will be propagated to the other
122 #[stable(feature = "rust1", since = "1.0.0")]
123 pub fn try_clone(&self) -> io::Result<TcpStream> {
124 self.0.duplicate().map(TcpStream)
127 /// Sets the read timeout to the timeout specified.
129 /// If the value specified is `None`, then `read` calls will block
130 /// indefinitely. It is an error to pass the zero `Duration` to this
135 /// Platforms may return a different error code whenever a read times out as
136 /// a result of setting this option. For example Unix typically returns an
137 /// error of the kind `WouldBlock`, but Windows may return `TimedOut`.
138 #[stable(feature = "socket_timeout", since = "1.4.0")]
139 pub fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
140 self.0.set_read_timeout(dur)
143 /// Sets the write timeout to the timeout specified.
145 /// If the value specified is `None`, then `write` calls will block
146 /// indefinitely. It is an error to pass the zero `Duration` to this
151 /// Platforms may return a different error code whenever a write times out
152 /// as a result of setting this option. For example Unix typically returns
153 /// an error of the kind `WouldBlock`, but Windows may return `TimedOut`.
154 #[stable(feature = "socket_timeout", since = "1.4.0")]
155 pub fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
156 self.0.set_write_timeout(dur)
159 /// Returns the read timeout of this socket.
161 /// If the timeout is `None`, then `read` calls will block indefinitely.
165 /// Some platforms do not provide access to the current timeout.
166 #[stable(feature = "socket_timeout", since = "1.4.0")]
167 pub fn read_timeout(&self) -> io::Result<Option<Duration>> {
168 self.0.read_timeout()
171 /// Returns the write timeout of this socket.
173 /// If the timeout is `None`, then `write` calls will block indefinitely.
177 /// Some platforms do not provide access to the current timeout.
178 #[stable(feature = "socket_timeout", since = "1.4.0")]
179 pub fn write_timeout(&self) -> io::Result<Option<Duration>> {
180 self.0.write_timeout()
183 /// Sets the value of the `TCP_NODELAY` option on this socket.
185 /// If set, this option disables the Nagle algorithm. This means that
186 /// segments are always sent as soon as possible, even if there is only a
187 /// small amount of data. When not set, data is buffered until there is a
188 /// sufficient amount to send out, thereby avoiding the frequent sending of
190 #[stable(feature = "net2_mutators", since = "1.9.0")]
191 pub fn set_nodelay(&self, nodelay: bool) -> io::Result<()> {
192 self.0.set_nodelay(nodelay)
195 /// Gets the value of the `TCP_NODELAY` option on this socket.
197 /// For more information about this option, see [`set_nodelay`][link].
199 /// [link]: #method.set_nodelay
200 #[stable(feature = "net2_mutators", since = "1.9.0")]
201 pub fn nodelay(&self) -> io::Result<bool> {
205 /// Sets the value for the `IP_TTL` option on this socket.
207 /// This value sets the time-to-live field that is used in every packet sent
208 /// from this socket.
209 #[stable(feature = "net2_mutators", since = "1.9.0")]
210 pub fn set_ttl(&self, ttl: u32) -> io::Result<()> {
214 /// Gets the value of the `IP_TTL` option for this socket.
216 /// For more information about this option, see [`set_ttl`][link].
218 /// [link]: #method.set_ttl
219 #[stable(feature = "net2_mutators", since = "1.9.0")]
220 pub fn ttl(&self) -> io::Result<u32> {
224 /// Get the value of the `SO_ERROR` option on this socket.
226 /// This will retrieve the stored error in the underlying socket, clearing
227 /// the field in the process. This can be useful for checking errors between
229 #[stable(feature = "net2_mutators", since = "1.9.0")]
230 pub fn take_error(&self) -> io::Result<Option<io::Error>> {
234 /// Moves this TCP stream into or out of nonblocking mode.
236 /// On Unix this corresponds to calling fcntl, and on Windows this
237 /// corresponds to calling ioctlsocket.
238 #[stable(feature = "net2_mutators", since = "1.9.0")]
239 pub fn set_nonblocking(&self, nonblocking: bool) -> io::Result<()> {
240 self.0.set_nonblocking(nonblocking)
244 #[stable(feature = "rust1", since = "1.0.0")]
245 impl Read for TcpStream {
246 fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> { self.0.read(buf) }
247 fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
248 self.0.read_to_end(buf)
251 #[stable(feature = "rust1", since = "1.0.0")]
252 impl Write for TcpStream {
253 fn write(&mut self, buf: &[u8]) -> io::Result<usize> { self.0.write(buf) }
254 fn flush(&mut self) -> io::Result<()> { Ok(()) }
256 #[stable(feature = "rust1", since = "1.0.0")]
257 impl<'a> Read for &'a TcpStream {
258 fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> { self.0.read(buf) }
259 fn read_to_end(&mut self, buf: &mut Vec<u8>) -> io::Result<usize> {
260 self.0.read_to_end(buf)
263 #[stable(feature = "rust1", since = "1.0.0")]
264 impl<'a> Write for &'a TcpStream {
265 fn write(&mut self, buf: &[u8]) -> io::Result<usize> { self.0.write(buf) }
266 fn flush(&mut self) -> io::Result<()> { Ok(()) }
269 impl AsInner<net_imp::TcpStream> for TcpStream {
270 fn as_inner(&self) -> &net_imp::TcpStream { &self.0 }
273 impl FromInner<net_imp::TcpStream> for TcpStream {
274 fn from_inner(inner: net_imp::TcpStream) -> TcpStream { TcpStream(inner) }
277 impl IntoInner<net_imp::TcpStream> for TcpStream {
278 fn into_inner(self) -> net_imp::TcpStream { self.0 }
281 #[stable(feature = "rust1", since = "1.0.0")]
282 impl fmt::Debug for TcpStream {
283 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
289 /// Creates a new `TcpListener` which will be bound to the specified
292 /// The returned listener is ready for accepting connections.
294 /// Binding with a port number of 0 will request that the OS assigns a port
295 /// to this listener. The port allocated can be queried via the
296 /// `local_addr` method.
298 /// The address type can be any implementor of `ToSocketAddrs` trait. See
299 /// its documentation for concrete examples.
300 #[stable(feature = "rust1", since = "1.0.0")]
301 pub fn bind<A: ToSocketAddrs>(addr: A) -> io::Result<TcpListener> {
302 super::each_addr(addr, net_imp::TcpListener::bind).map(TcpListener)
305 /// Returns the local socket address of this listener.
306 #[stable(feature = "rust1", since = "1.0.0")]
307 pub fn local_addr(&self) -> io::Result<SocketAddr> {
311 /// Creates a new independently owned handle to the underlying socket.
313 /// The returned `TcpListener` is a reference to the same socket that this
314 /// object references. Both handles can be used to accept incoming
315 /// connections and options set on one listener will affect the other.
316 #[stable(feature = "rust1", since = "1.0.0")]
317 pub fn try_clone(&self) -> io::Result<TcpListener> {
318 self.0.duplicate().map(TcpListener)
321 /// Accept a new incoming connection from this listener.
323 /// This function will block the calling thread until a new TCP connection
324 /// is established. When established, the corresponding `TcpStream` and the
325 /// remote peer's address will be returned.
326 #[stable(feature = "rust1", since = "1.0.0")]
327 pub fn accept(&self) -> io::Result<(TcpStream, SocketAddr)> {
328 self.0.accept().map(|(a, b)| (TcpStream(a), b))
331 /// Returns an iterator over the connections being received on this
334 /// The returned iterator will never return `None` and will also not yield
335 /// the peer's `SocketAddr` structure.
336 #[stable(feature = "rust1", since = "1.0.0")]
337 pub fn incoming(&self) -> Incoming {
338 Incoming { listener: self }
341 /// Sets the value for the `IP_TTL` option on this socket.
343 /// This value sets the time-to-live field that is used in every packet sent
344 /// from this socket.
345 #[stable(feature = "net2_mutators", since = "1.9.0")]
346 pub fn set_ttl(&self, ttl: u32) -> io::Result<()> {
350 /// Gets the value of the `IP_TTL` option for this socket.
352 /// For more information about this option, see [`set_ttl`][link].
354 /// [link]: #method.set_ttl
355 #[stable(feature = "net2_mutators", since = "1.9.0")]
356 pub fn ttl(&self) -> io::Result<u32> {
360 /// Sets the value for the `IPV6_V6ONLY` option on this socket.
362 /// If this is set to `true` then the socket is restricted to sending and
363 /// receiving IPv6 packets only. In this case two IPv4 and IPv6 applications
364 /// can bind the same port at the same time.
366 /// If this is set to `false` then the socket can be used to send and
367 /// receive packets from an IPv4-mapped IPv6 address.
368 #[stable(feature = "net2_mutators", since = "1.9.0")]
369 pub fn set_only_v6(&self, only_v6: bool) -> io::Result<()> {
370 self.0.set_only_v6(only_v6)
373 /// Gets the value of the `IPV6_V6ONLY` option for this socket.
375 /// For more information about this option, see [`set_only_v6`][link].
377 /// [link]: #method.set_only_v6
378 #[stable(feature = "net2_mutators", since = "1.9.0")]
379 pub fn only_v6(&self) -> io::Result<bool> {
383 /// Get the value of the `SO_ERROR` option on this socket.
385 /// This will retrieve the stored error in the underlying socket, clearing
386 /// the field in the process. This can be useful for checking errors between
388 #[stable(feature = "net2_mutators", since = "1.9.0")]
389 pub fn take_error(&self) -> io::Result<Option<io::Error>> {
393 /// Moves this TCP stream into or out of nonblocking mode.
395 /// On Unix this corresponds to calling fcntl, and on Windows this
396 /// corresponds to calling ioctlsocket.
397 #[stable(feature = "net2_mutators", since = "1.9.0")]
398 pub fn set_nonblocking(&self, nonblocking: bool) -> io::Result<()> {
399 self.0.set_nonblocking(nonblocking)
403 #[stable(feature = "rust1", since = "1.0.0")]
404 impl<'a> Iterator for Incoming<'a> {
405 type Item = io::Result<TcpStream>;
406 fn next(&mut self) -> Option<io::Result<TcpStream>> {
407 Some(self.listener.accept().map(|p| p.0))
411 impl AsInner<net_imp::TcpListener> for TcpListener {
412 fn as_inner(&self) -> &net_imp::TcpListener { &self.0 }
415 impl FromInner<net_imp::TcpListener> for TcpListener {
416 fn from_inner(inner: net_imp::TcpListener) -> TcpListener {
421 impl IntoInner<net_imp::TcpListener> for TcpListener {
422 fn into_inner(self) -> net_imp::TcpListener { self.0 }
425 #[stable(feature = "rust1", since = "1.0.0")]
426 impl fmt::Debug for TcpListener {
427 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
439 use net::test::{next_test_ip4, next_test_ip6};
440 use sync::mpsc::channel;
441 use sys_common::AsInner;
442 use time::{Instant, Duration};
445 fn each_ip(f: &mut FnMut(SocketAddr)) {
454 Err(e) => panic!("received error for `{}`: {}", stringify!($e), e),
461 match TcpListener::bind("1.1.1.1:9999") {
464 assert_eq!(e.kind(), ErrorKind::AddrNotAvailable),
470 match TcpStream::connect("0.0.0.0:1") {
472 Err(e) => assert!(e.kind() == ErrorKind::ConnectionRefused ||
473 e.kind() == ErrorKind::InvalidInput ||
474 e.kind() == ErrorKind::AddrInUse ||
475 e.kind() == ErrorKind::AddrNotAvailable,
476 "bad error: {} {:?}", e, e.kind()),
481 fn listen_localhost() {
482 let socket_addr = next_test_ip4();
483 let listener = t!(TcpListener::bind(&socket_addr));
485 let _t = thread::spawn(move || {
486 let mut stream = t!(TcpStream::connect(&("localhost",
487 socket_addr.port())));
488 t!(stream.write(&[144]));
491 let mut stream = t!(listener.accept()).0;
493 t!(stream.read(&mut buf));
494 assert!(buf[0] == 144);
498 fn connect_loopback() {
499 each_ip(&mut |addr| {
500 let acceptor = t!(TcpListener::bind(&addr));
502 let _t = thread::spawn(move|| {
503 let host = match addr {
504 SocketAddr::V4(..) => "127.0.0.1",
505 SocketAddr::V6(..) => "::1",
507 let mut stream = t!(TcpStream::connect(&(host, addr.port())));
508 t!(stream.write(&[66]));
511 let mut stream = t!(acceptor.accept()).0;
513 t!(stream.read(&mut buf));
514 assert!(buf[0] == 66);
520 each_ip(&mut |addr| {
521 let acceptor = t!(TcpListener::bind(&addr));
523 let (tx, rx) = channel();
524 let _t = thread::spawn(move|| {
525 let mut stream = t!(TcpStream::connect(&addr));
526 t!(stream.write(&[99]));
527 tx.send(t!(stream.local_addr())).unwrap();
530 let (mut stream, addr) = t!(acceptor.accept());
532 t!(stream.read(&mut buf));
533 assert!(buf[0] == 99);
534 assert_eq!(addr, t!(rx.recv()));
540 each_ip(&mut |addr| {
541 let acceptor = t!(TcpListener::bind(&addr));
543 let _t = thread::spawn(move|| {
544 let _stream = t!(TcpStream::connect(&addr));
548 let mut stream = t!(acceptor.accept()).0;
550 let nread = t!(stream.read(&mut buf));
551 assert_eq!(nread, 0);
552 let nread = t!(stream.read(&mut buf));
553 assert_eq!(nread, 0);
559 each_ip(&mut |addr| {
560 let acceptor = t!(TcpListener::bind(&addr));
562 let (tx, rx) = channel();
563 let _t = thread::spawn(move|| {
564 drop(t!(TcpStream::connect(&addr)));
565 tx.send(()).unwrap();
568 let mut stream = t!(acceptor.accept()).0;
571 match stream.write(&buf) {
574 assert!(e.kind() == ErrorKind::ConnectionReset ||
575 e.kind() == ErrorKind::BrokenPipe ||
576 e.kind() == ErrorKind::ConnectionAborted,
577 "unknown error: {}", e);
584 fn multiple_connect_serial() {
585 each_ip(&mut |addr| {
587 let acceptor = t!(TcpListener::bind(&addr));
589 let _t = thread::spawn(move|| {
591 let mut stream = t!(TcpStream::connect(&addr));
592 t!(stream.write(&[99]));
596 for stream in acceptor.incoming().take(max) {
597 let mut stream = t!(stream);
599 t!(stream.read(&mut buf));
600 assert_eq!(buf[0], 99);
606 fn multiple_connect_interleaved_greedy_schedule() {
607 const MAX: usize = 10;
608 each_ip(&mut |addr| {
609 let acceptor = t!(TcpListener::bind(&addr));
611 let _t = thread::spawn(move|| {
612 let acceptor = acceptor;
613 for (i, stream) in acceptor.incoming().enumerate().take(MAX) {
614 // Start another thread to handle the connection
615 let _t = thread::spawn(move|| {
616 let mut stream = t!(stream);
618 t!(stream.read(&mut buf));
619 assert!(buf[0] == i as u8);
627 fn connect(i: usize, addr: SocketAddr) {
628 if i == MAX { return }
630 let t = thread::spawn(move|| {
631 let mut stream = t!(TcpStream::connect(&addr));
632 // Connect again before writing
633 connect(i + 1, addr);
634 t!(stream.write(&[i as u8]));
636 t.join().ok().unwrap();
641 fn multiple_connect_interleaved_lazy_schedule() {
642 const MAX: usize = 10;
643 each_ip(&mut |addr| {
644 let acceptor = t!(TcpListener::bind(&addr));
646 let _t = thread::spawn(move|| {
647 for stream in acceptor.incoming().take(MAX) {
648 // Start another thread to handle the connection
649 let _t = thread::spawn(move|| {
650 let mut stream = t!(stream);
652 t!(stream.read(&mut buf));
653 assert!(buf[0] == 99);
661 fn connect(i: usize, addr: SocketAddr) {
662 if i == MAX { return }
664 let t = thread::spawn(move|| {
665 let mut stream = t!(TcpStream::connect(&addr));
666 connect(i + 1, addr);
667 t!(stream.write(&[99]));
669 t.join().ok().unwrap();
674 fn socket_and_peer_name() {
675 each_ip(&mut |addr| {
676 let listener = t!(TcpListener::bind(&addr));
677 let so_name = t!(listener.local_addr());
678 assert_eq!(addr, so_name);
679 let _t = thread::spawn(move|| {
680 t!(listener.accept());
683 let stream = t!(TcpStream::connect(&addr));
684 assert_eq!(addr, t!(stream.peer_addr()));
690 each_ip(&mut |addr| {
691 let (tx, rx) = channel();
692 let srv = t!(TcpListener::bind(&addr));
693 let _t = thread::spawn(move|| {
694 let mut cl = t!(srv.accept()).0;
695 cl.write(&[10]).unwrap();
698 tx.send(()).unwrap();
701 let mut c = t!(TcpStream::connect(&addr));
703 assert_eq!(c.read(&mut b).unwrap(), 1);
711 each_ip(&mut |addr| {
712 let _listener = t!(TcpListener::bind(&addr));
713 match TcpListener::bind(&addr) {
716 assert!(e.kind() == ErrorKind::ConnectionRefused ||
717 e.kind() == ErrorKind::Other ||
718 e.kind() == ErrorKind::AddrInUse,
719 "unknown error: {} {:?}", e, e.kind());
727 each_ip(&mut |addr| {
728 let acceptor = t!(TcpListener::bind(&addr));
730 let _t = thread::spawn(move|| {
731 t!(TcpStream::connect(&addr));
734 t!(acceptor.accept());
736 t!(TcpListener::bind(&addr));
741 fn tcp_clone_smoke() {
742 each_ip(&mut |addr| {
743 let acceptor = t!(TcpListener::bind(&addr));
745 let _t = thread::spawn(move|| {
746 let mut s = t!(TcpStream::connect(&addr));
747 let mut buf = [0, 0];
748 assert_eq!(s.read(&mut buf).unwrap(), 1);
749 assert_eq!(buf[0], 1);
753 let mut s1 = t!(acceptor.accept()).0;
754 let s2 = t!(s1.try_clone());
756 let (tx1, rx1) = channel();
757 let (tx2, rx2) = channel();
758 let _t = thread::spawn(move|| {
762 tx2.send(()).unwrap();
764 tx1.send(()).unwrap();
765 let mut buf = [0, 0];
766 assert_eq!(s1.read(&mut buf).unwrap(), 1);
772 fn tcp_clone_two_read() {
773 each_ip(&mut |addr| {
774 let acceptor = t!(TcpListener::bind(&addr));
775 let (tx1, rx) = channel();
776 let tx2 = tx1.clone();
778 let _t = thread::spawn(move|| {
779 let mut s = t!(TcpStream::connect(&addr));
786 let mut s1 = t!(acceptor.accept()).0;
787 let s2 = t!(s1.try_clone());
789 let (done, rx) = channel();
790 let _t = thread::spawn(move|| {
792 let mut buf = [0, 0];
793 t!(s2.read(&mut buf));
794 tx2.send(()).unwrap();
795 done.send(()).unwrap();
797 let mut buf = [0, 0];
798 t!(s1.read(&mut buf));
799 tx1.send(()).unwrap();
806 fn tcp_clone_two_write() {
807 each_ip(&mut |addr| {
808 let acceptor = t!(TcpListener::bind(&addr));
810 let _t = thread::spawn(move|| {
811 let mut s = t!(TcpStream::connect(&addr));
812 let mut buf = [0, 1];
813 t!(s.read(&mut buf));
814 t!(s.read(&mut buf));
817 let mut s1 = t!(acceptor.accept()).0;
818 let s2 = t!(s1.try_clone());
820 let (done, rx) = channel();
821 let _t = thread::spawn(move|| {
824 done.send(()).unwrap();
833 fn shutdown_smoke() {
834 each_ip(&mut |addr| {
835 let a = t!(TcpListener::bind(&addr));
836 let _t = thread::spawn(move|| {
837 let mut c = t!(a.accept()).0;
839 assert_eq!(c.read(&mut b).unwrap(), 0);
843 let mut s = t!(TcpStream::connect(&addr));
844 t!(s.shutdown(Shutdown::Write));
845 assert!(s.write(&[1]).is_err());
847 assert_eq!(t!(s.read(&mut b)), 1);
853 fn close_readwrite_smoke() {
854 each_ip(&mut |addr| {
855 let a = t!(TcpListener::bind(&addr));
856 let (tx, rx) = channel::<()>();
857 let _t = thread::spawn(move|| {
858 let _s = t!(a.accept());
863 let mut s = t!(TcpStream::connect(&addr));
864 let mut s2 = t!(s.try_clone());
866 // closing should prevent reads/writes
867 t!(s.shutdown(Shutdown::Write));
868 assert!(s.write(&[0]).is_err());
869 t!(s.shutdown(Shutdown::Read));
870 assert_eq!(s.read(&mut b).unwrap(), 0);
872 // closing should affect previous handles
873 assert!(s2.write(&[0]).is_err());
874 assert_eq!(s2.read(&mut b).unwrap(), 0);
876 // closing should affect new handles
877 let mut s3 = t!(s.try_clone());
878 assert!(s3.write(&[0]).is_err());
879 assert_eq!(s3.read(&mut b).unwrap(), 0);
881 // make sure these don't die
882 let _ = s2.shutdown(Shutdown::Read);
883 let _ = s2.shutdown(Shutdown::Write);
884 let _ = s3.shutdown(Shutdown::Read);
885 let _ = s3.shutdown(Shutdown::Write);
891 fn close_read_wakes_up() {
892 each_ip(&mut |addr| {
893 let a = t!(TcpListener::bind(&addr));
894 let (tx1, rx) = channel::<()>();
895 let _t = thread::spawn(move|| {
896 let _s = t!(a.accept());
900 let s = t!(TcpStream::connect(&addr));
901 let s2 = t!(s.try_clone());
902 let (tx, rx) = channel();
903 let _t = thread::spawn(move|| {
905 assert_eq!(t!(s2.read(&mut [0])), 0);
906 tx.send(()).unwrap();
908 // this should wake up the child thread
909 t!(s.shutdown(Shutdown::Read));
911 // this test will never finish if the child doesn't wake up
918 fn clone_while_reading() {
919 each_ip(&mut |addr| {
920 let accept = t!(TcpListener::bind(&addr));
922 // Enqueue a thread to write to a socket
923 let (tx, rx) = channel();
924 let (txdone, rxdone) = channel();
925 let txdone2 = txdone.clone();
926 let _t = thread::spawn(move|| {
927 let mut tcp = t!(TcpStream::connect(&addr));
930 txdone2.send(()).unwrap();
933 // Spawn off a reading clone
934 let tcp = t!(accept.accept()).0;
935 let tcp2 = t!(tcp.try_clone());
936 let txdone3 = txdone.clone();
937 let _t = thread::spawn(move|| {
939 t!(tcp2.read(&mut [0]));
940 txdone3.send(()).unwrap();
943 // Try to ensure that the reading clone is indeed reading
948 // clone the handle again while it's reading, then let it finish the
950 let _ = t!(tcp.try_clone());
951 tx.send(()).unwrap();
952 rxdone.recv().unwrap();
953 rxdone.recv().unwrap();
958 fn clone_accept_smoke() {
959 each_ip(&mut |addr| {
960 let a = t!(TcpListener::bind(&addr));
961 let a2 = t!(a.try_clone());
963 let _t = thread::spawn(move|| {
964 let _ = TcpStream::connect(&addr);
966 let _t = thread::spawn(move|| {
967 let _ = TcpStream::connect(&addr);
976 fn clone_accept_concurrent() {
977 each_ip(&mut |addr| {
978 let a = t!(TcpListener::bind(&addr));
979 let a2 = t!(a.try_clone());
981 let (tx, rx) = channel();
982 let tx2 = tx.clone();
984 let _t = thread::spawn(move|| {
985 tx.send(t!(a.accept())).unwrap();
987 let _t = thread::spawn(move|| {
988 tx2.send(t!(a2.accept())).unwrap();
991 let _t = thread::spawn(move|| {
992 let _ = TcpStream::connect(&addr);
994 let _t = thread::spawn(move|| {
995 let _ = TcpStream::connect(&addr);
1005 let name = if cfg!(windows) {"socket"} else {"fd"};
1006 let socket_addr = next_test_ip4();
1008 let listener = t!(TcpListener::bind(&socket_addr));
1009 let listener_inner = listener.0.socket().as_inner();
1010 let compare = format!("TcpListener {{ addr: {:?}, {}: {:?} }}",
1011 socket_addr, name, listener_inner);
1012 assert_eq!(format!("{:?}", listener), compare);
1014 let stream = t!(TcpStream::connect(&("localhost",
1015 socket_addr.port())));
1016 let stream_inner = stream.0.socket().as_inner();
1017 let compare = format!("TcpStream {{ addr: {:?}, \
1018 peer: {:?}, {}: {:?} }}",
1019 stream.local_addr().unwrap(),
1020 stream.peer_addr().unwrap(),
1023 assert_eq!(format!("{:?}", stream), compare);
1026 // FIXME: re-enabled bitrig/openbsd tests once their socket timeout code
1027 // no longer has rounding errors.
1028 #[cfg_attr(any(target_os = "bitrig", target_os = "netbsd", target_os = "openbsd"), ignore)]
1031 let addr = next_test_ip4();
1032 let listener = t!(TcpListener::bind(&addr));
1034 let stream = t!(TcpStream::connect(&("localhost", addr.port())));
1035 let dur = Duration::new(15410, 0);
1037 assert_eq!(None, t!(stream.read_timeout()));
1039 t!(stream.set_read_timeout(Some(dur)));
1040 assert_eq!(Some(dur), t!(stream.read_timeout()));
1042 assert_eq!(None, t!(stream.write_timeout()));
1044 t!(stream.set_write_timeout(Some(dur)));
1045 assert_eq!(Some(dur), t!(stream.write_timeout()));
1047 t!(stream.set_read_timeout(None));
1048 assert_eq!(None, t!(stream.read_timeout()));
1050 t!(stream.set_write_timeout(None));
1051 assert_eq!(None, t!(stream.write_timeout()));
1056 fn test_read_timeout() {
1057 let addr = next_test_ip4();
1058 let listener = t!(TcpListener::bind(&addr));
1060 let mut stream = t!(TcpStream::connect(&("localhost", addr.port())));
1061 t!(stream.set_read_timeout(Some(Duration::from_millis(1000))));
1063 let mut buf = [0; 10];
1064 let start = Instant::now();
1065 let kind = stream.read(&mut buf).err().expect("expected error").kind();
1066 assert!(kind == ErrorKind::WouldBlock || kind == ErrorKind::TimedOut);
1067 assert!(start.elapsed() > Duration::from_millis(400));
1072 fn test_read_with_timeout() {
1073 let addr = next_test_ip4();
1074 let listener = t!(TcpListener::bind(&addr));
1076 let mut stream = t!(TcpStream::connect(&("localhost", addr.port())));
1077 t!(stream.set_read_timeout(Some(Duration::from_millis(1000))));
1079 let mut other_end = t!(listener.accept()).0;
1080 t!(other_end.write_all(b"hello world"));
1082 let mut buf = [0; 11];
1083 t!(stream.read(&mut buf));
1084 assert_eq!(b"hello world", &buf[..]);
1086 let start = Instant::now();
1087 let kind = stream.read(&mut buf).err().expect("expected error").kind();
1088 assert!(kind == ErrorKind::WouldBlock || kind == ErrorKind::TimedOut);
1089 assert!(start.elapsed() > Duration::from_millis(400));
1095 let addr = next_test_ip4();
1096 let _listener = t!(TcpListener::bind(&addr));
1098 let stream = t!(TcpStream::connect(&("localhost", addr.port())));
1100 assert_eq!(false, t!(stream.nodelay()));
1101 t!(stream.set_nodelay(true));
1102 assert_eq!(true, t!(stream.nodelay()));
1103 t!(stream.set_nodelay(false));
1104 assert_eq!(false, t!(stream.nodelay()));
1111 let addr = next_test_ip4();
1112 let listener = t!(TcpListener::bind(&addr));
1114 t!(listener.set_ttl(ttl));
1115 assert_eq!(ttl, t!(listener.ttl()));
1117 let stream = t!(TcpStream::connect(&("localhost", addr.port())));
1119 t!(stream.set_ttl(ttl));
1120 assert_eq!(ttl, t!(stream.ttl()));
1124 fn set_nonblocking() {
1125 let addr = next_test_ip4();
1126 let listener = t!(TcpListener::bind(&addr));
1128 t!(listener.set_nonblocking(true));
1129 t!(listener.set_nonblocking(false));
1131 let mut stream = t!(TcpStream::connect(&("localhost", addr.port())));
1133 t!(stream.set_nonblocking(false));
1134 t!(stream.set_nonblocking(true));
1137 match stream.read(&mut buf) {
1138 Ok(_) => panic!("expected error"),
1139 Err(ref e) if e.kind() == ErrorKind::WouldBlock => {}
1140 Err(e) => panic!("unexpected error {}", e),