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()
185 #[stable(feature = "rust1", since = "1.0.0")]
186 impl Read
for TcpStream
{
187 fn read(&mut self, buf
: &mut [u8]) -> io
::Result
<usize> { self.0.read(buf) }
188 fn read_to_end(&mut self, buf
: &mut Vec
<u8>) -> io
::Result
<usize> {
189 unsafe { read_to_end_uninitialized(self, buf) }
192 #[stable(feature = "rust1", since = "1.0.0")]
193 impl Write
for TcpStream
{
194 fn write(&mut self, buf
: &[u8]) -> io
::Result
<usize> { self.0.write(buf) }
195 fn flush(&mut self) -> io
::Result
<()> { Ok(()) }
197 #[stable(feature = "rust1", since = "1.0.0")]
198 impl<'a
> Read
for &'a TcpStream
{
199 fn read(&mut self, buf
: &mut [u8]) -> io
::Result
<usize> { self.0.read(buf) }
200 fn read_to_end(&mut self, buf
: &mut Vec
<u8>) -> io
::Result
<usize> {
201 unsafe { read_to_end_uninitialized(self, buf) }
204 #[stable(feature = "rust1", since = "1.0.0")]
205 impl<'a
> Write
for &'a TcpStream
{
206 fn write(&mut self, buf
: &[u8]) -> io
::Result
<usize> { self.0.write(buf) }
207 fn flush(&mut self) -> io
::Result
<()> { Ok(()) }
210 impl AsInner
<net_imp
::TcpStream
> for TcpStream
{
211 fn as_inner(&self) -> &net_imp
::TcpStream { &self.0 }
214 impl FromInner
<net_imp
::TcpStream
> for TcpStream
{
215 fn from_inner(inner
: net_imp
::TcpStream
) -> TcpStream { TcpStream(inner) }
218 impl IntoInner
<net_imp
::TcpStream
> for TcpStream
{
219 fn into_inner(self) -> net_imp
::TcpStream { self.0 }
222 #[stable(feature = "rust1", since = "1.0.0")]
223 impl fmt
::Debug
for TcpStream
{
224 fn fmt(&self, f
: &mut fmt
::Formatter
) -> fmt
::Result
{
230 /// Creates a new `TcpListener` which will be bound to the specified
233 /// The returned listener is ready for accepting connections.
235 /// Binding with a port number of 0 will request that the OS assigns a port
236 /// to this listener. The port allocated can be queried via the
237 /// `local_addr` method.
239 /// The address type can be any implementor of `ToSocketAddrs` trait. See
240 /// its documentation for concrete examples.
241 #[stable(feature = "rust1", since = "1.0.0")]
242 pub fn bind
<A
: ToSocketAddrs
>(addr
: A
) -> io
::Result
<TcpListener
> {
243 super::each_addr(addr
, net_imp
::TcpListener
::bind
).map(TcpListener
)
246 /// Returns the local socket address of this listener.
247 #[stable(feature = "rust1", since = "1.0.0")]
248 pub fn local_addr(&self) -> io
::Result
<SocketAddr
> {
252 /// Creates a new independently owned handle to the underlying socket.
254 /// The returned `TcpListener` is a reference to the same socket that this
255 /// object references. Both handles can be used to accept incoming
256 /// connections and options set on one listener will affect the other.
257 #[stable(feature = "rust1", since = "1.0.0")]
258 pub fn try_clone(&self) -> io
::Result
<TcpListener
> {
259 self.0.duplicate().map(TcpListener
)
262 /// Accept a new incoming connection from this listener.
264 /// This function will block the calling thread until a new TCP connection
265 /// is established. When established, the corresponding `TcpStream` and the
266 /// remote peer's address will be returned.
267 #[stable(feature = "rust1", since = "1.0.0")]
268 pub fn accept(&self) -> io
::Result
<(TcpStream
, SocketAddr
)> {
269 self.0.accept().map(|(a
, b
)| (TcpStream(a
), b
))
272 /// Returns an iterator over the connections being received on this
275 /// The returned iterator will never return `None` and will also not yield
276 /// the peer's `SocketAddr` structure.
277 #[stable(feature = "rust1", since = "1.0.0")]
278 pub fn incoming(&self) -> Incoming
{
279 Incoming { listener: self }
283 #[stable(feature = "rust1", since = "1.0.0")]
284 impl<'a
> Iterator
for Incoming
<'a
> {
285 type Item
= io
::Result
<TcpStream
>;
286 fn next(&mut self) -> Option
<io
::Result
<TcpStream
>> {
287 Some(self.listener
.accept().map(|p
| p
.0))
291 impl AsInner
<net_imp
::TcpListener
> for TcpListener
{
292 fn as_inner(&self) -> &net_imp
::TcpListener { &self.0 }
295 impl FromInner
<net_imp
::TcpListener
> for TcpListener
{
296 fn from_inner(inner
: net_imp
::TcpListener
) -> TcpListener
{
301 impl IntoInner
<net_imp
::TcpListener
> for TcpListener
{
302 fn into_inner(self) -> net_imp
::TcpListener { self.0 }
305 #[stable(feature = "rust1", since = "1.0.0")]
306 impl fmt
::Debug
for TcpListener
{
307 fn fmt(&self, f
: &mut fmt
::Formatter
) -> fmt
::Result
{
319 use net
::test
::{next_test_ip4, next_test_ip6}
;
320 use sync
::mpsc
::channel
;
321 use sys_common
::AsInner
;
325 fn each_ip(f
: &mut FnMut(SocketAddr
)) {
334 Err(e
) => panic
!("received error for `{}`: {}", stringify
!($e
), e
),
341 match TcpListener
::bind("1.1.1.1:9999") {
344 assert_eq
!(e
.kind(), ErrorKind
::AddrNotAvailable
),
350 match TcpStream
::connect("0.0.0.0:1") {
352 Err(e
) => assert
!(e
.kind() == ErrorKind
::ConnectionRefused
||
353 e
.kind() == ErrorKind
::InvalidInput
||
354 e
.kind() == ErrorKind
::AddrInUse
||
355 e
.kind() == ErrorKind
::AddrNotAvailable
,
356 "bad error: {} {:?}", e
, e
.kind()),
361 fn listen_localhost() {
362 let socket_addr
= next_test_ip4();
363 let listener
= t
!(TcpListener
::bind(&socket_addr
));
365 let _t
= thread
::spawn(move || {
366 let mut stream
= t
!(TcpStream
::connect(&("localhost",
367 socket_addr
.port())));
368 t
!(stream
.write(&[144]));
371 let mut stream
= t
!(listener
.accept()).0;
373 t
!(stream
.read(&mut buf
));
374 assert
!(buf
[0] == 144);
378 fn connect_loopback() {
379 each_ip(&mut |addr
| {
380 let acceptor
= t
!(TcpListener
::bind(&addr
));
382 let _t
= thread
::spawn(move|| {
383 let host
= match addr
{
384 SocketAddr
::V4(..) => "127.0.0.1",
385 SocketAddr
::V6(..) => "::1",
387 let mut stream
= t
!(TcpStream
::connect(&(host
, addr
.port())));
388 t
!(stream
.write(&[66]));
391 let mut stream
= t
!(acceptor
.accept()).0;
393 t
!(stream
.read(&mut buf
));
394 assert
!(buf
[0] == 66);
400 each_ip(&mut |addr
| {
401 let acceptor
= t
!(TcpListener
::bind(&addr
));
403 let (tx
, rx
) = channel();
404 let _t
= thread
::spawn(move|| {
405 let mut stream
= t
!(TcpStream
::connect(&addr
));
406 t
!(stream
.write(&[99]));
407 tx
.send(t
!(stream
.local_addr())).unwrap();
410 let (mut stream
, addr
) = t
!(acceptor
.accept());
412 t
!(stream
.read(&mut buf
));
413 assert
!(buf
[0] == 99);
414 assert_eq
!(addr
, t
!(rx
.recv()));
420 each_ip(&mut |addr
| {
421 let acceptor
= t
!(TcpListener
::bind(&addr
));
423 let _t
= thread
::spawn(move|| {
424 let _stream
= t
!(TcpStream
::connect(&addr
));
428 let mut stream
= t
!(acceptor
.accept()).0;
430 let nread
= t
!(stream
.read(&mut buf
));
431 assert_eq
!(nread
, 0);
432 let nread
= t
!(stream
.read(&mut buf
));
433 assert_eq
!(nread
, 0);
439 each_ip(&mut |addr
| {
440 let acceptor
= t
!(TcpListener
::bind(&addr
));
442 let (tx
, rx
) = channel();
443 let _t
= thread
::spawn(move|| {
444 drop(t
!(TcpStream
::connect(&addr
)));
445 tx
.send(()).unwrap();
448 let mut stream
= t
!(acceptor
.accept()).0;
451 match stream
.write(&buf
) {
454 assert
!(e
.kind() == ErrorKind
::ConnectionReset
||
455 e
.kind() == ErrorKind
::BrokenPipe
||
456 e
.kind() == ErrorKind
::ConnectionAborted
,
457 "unknown error: {}", e
);
464 fn multiple_connect_serial() {
465 each_ip(&mut |addr
| {
467 let acceptor
= t
!(TcpListener
::bind(&addr
));
469 let _t
= thread
::spawn(move|| {
471 let mut stream
= t
!(TcpStream
::connect(&addr
));
472 t
!(stream
.write(&[99]));
476 for stream
in acceptor
.incoming().take(max
) {
477 let mut stream
= t
!(stream
);
479 t
!(stream
.read(&mut buf
));
480 assert_eq
!(buf
[0], 99);
486 fn multiple_connect_interleaved_greedy_schedule() {
487 const MAX
: usize = 10;
488 each_ip(&mut |addr
| {
489 let acceptor
= t
!(TcpListener
::bind(&addr
));
491 let _t
= thread
::spawn(move|| {
492 let acceptor
= acceptor
;
493 for (i
, stream
) in acceptor
.incoming().enumerate().take(MAX
) {
494 // Start another thread to handle the connection
495 let _t
= thread
::spawn(move|| {
496 let mut stream
= t
!(stream
);
498 t
!(stream
.read(&mut buf
));
499 assert
!(buf
[0] == i
as u8);
507 fn connect(i
: usize, addr
: SocketAddr
) {
508 if i
== MAX { return }
510 let t
= thread
::spawn(move|| {
511 let mut stream
= t
!(TcpStream
::connect(&addr
));
512 // Connect again before writing
513 connect(i
+ 1, addr
);
514 t
!(stream
.write(&[i
as u8]));
516 t
.join().ok().unwrap();
521 fn multiple_connect_interleaved_lazy_schedule() {
522 const MAX
: usize = 10;
523 each_ip(&mut |addr
| {
524 let acceptor
= t
!(TcpListener
::bind(&addr
));
526 let _t
= thread
::spawn(move|| {
527 for stream
in acceptor
.incoming().take(MAX
) {
528 // Start another thread to handle the connection
529 let _t
= thread
::spawn(move|| {
530 let mut stream
= t
!(stream
);
532 t
!(stream
.read(&mut buf
));
533 assert
!(buf
[0] == 99);
541 fn connect(i
: usize, addr
: SocketAddr
) {
542 if i
== MAX { return }
544 let t
= thread
::spawn(move|| {
545 let mut stream
= t
!(TcpStream
::connect(&addr
));
546 connect(i
+ 1, addr
);
547 t
!(stream
.write(&[99]));
549 t
.join().ok().unwrap();
554 fn socket_and_peer_name() {
555 each_ip(&mut |addr
| {
556 let listener
= t
!(TcpListener
::bind(&addr
));
557 let so_name
= t
!(listener
.local_addr());
558 assert_eq
!(addr
, so_name
);
559 let _t
= thread
::spawn(move|| {
560 t
!(listener
.accept());
563 let stream
= t
!(TcpStream
::connect(&addr
));
564 assert_eq
!(addr
, t
!(stream
.peer_addr()));
570 each_ip(&mut |addr
| {
571 let (tx
, rx
) = channel();
572 let srv
= t
!(TcpListener
::bind(&addr
));
573 let _t
= thread
::spawn(move|| {
574 let mut cl
= t
!(srv
.accept()).0;
575 cl
.write(&[10]).unwrap();
578 tx
.send(()).unwrap();
581 let mut c
= t
!(TcpStream
::connect(&addr
));
583 assert_eq
!(c
.read(&mut b
).unwrap(), 1);
591 each_ip(&mut |addr
| {
592 let _listener
= t
!(TcpListener
::bind(&addr
));
593 match TcpListener
::bind(&addr
) {
596 assert
!(e
.kind() == ErrorKind
::ConnectionRefused
||
597 e
.kind() == ErrorKind
::Other
||
598 e
.kind() == ErrorKind
::AddrInUse
,
599 "unknown error: {} {:?}", e
, e
.kind());
607 each_ip(&mut |addr
| {
608 let acceptor
= t
!(TcpListener
::bind(&addr
));
610 let _t
= thread
::spawn(move|| {
611 t
!(TcpStream
::connect(&addr
));
614 t
!(acceptor
.accept());
616 t
!(TcpListener
::bind(&addr
));
621 fn tcp_clone_smoke() {
622 each_ip(&mut |addr
| {
623 let acceptor
= t
!(TcpListener
::bind(&addr
));
625 let _t
= thread
::spawn(move|| {
626 let mut s
= t
!(TcpStream
::connect(&addr
));
627 let mut buf
= [0, 0];
628 assert_eq
!(s
.read(&mut buf
).unwrap(), 1);
629 assert_eq
!(buf
[0], 1);
633 let mut s1
= t
!(acceptor
.accept()).0;
634 let s2
= t
!(s1
.try_clone());
636 let (tx1
, rx1
) = channel();
637 let (tx2
, rx2
) = channel();
638 let _t
= thread
::spawn(move|| {
642 tx2
.send(()).unwrap();
644 tx1
.send(()).unwrap();
645 let mut buf
= [0, 0];
646 assert_eq
!(s1
.read(&mut buf
).unwrap(), 1);
652 fn tcp_clone_two_read() {
653 each_ip(&mut |addr
| {
654 let acceptor
= t
!(TcpListener
::bind(&addr
));
655 let (tx1
, rx
) = channel();
656 let tx2
= tx1
.clone();
658 let _t
= thread
::spawn(move|| {
659 let mut s
= t
!(TcpStream
::connect(&addr
));
666 let mut s1
= t
!(acceptor
.accept()).0;
667 let s2
= t
!(s1
.try_clone());
669 let (done
, rx
) = channel();
670 let _t
= thread
::spawn(move|| {
672 let mut buf
= [0, 0];
673 t
!(s2
.read(&mut buf
));
674 tx2
.send(()).unwrap();
675 done
.send(()).unwrap();
677 let mut buf
= [0, 0];
678 t
!(s1
.read(&mut buf
));
679 tx1
.send(()).unwrap();
686 fn tcp_clone_two_write() {
687 each_ip(&mut |addr
| {
688 let acceptor
= t
!(TcpListener
::bind(&addr
));
690 let _t
= thread
::spawn(move|| {
691 let mut s
= t
!(TcpStream
::connect(&addr
));
692 let mut buf
= [0, 1];
693 t
!(s
.read(&mut buf
));
694 t
!(s
.read(&mut buf
));
697 let mut s1
= t
!(acceptor
.accept()).0;
698 let s2
= t
!(s1
.try_clone());
700 let (done
, rx
) = channel();
701 let _t
= thread
::spawn(move|| {
704 done
.send(()).unwrap();
713 fn shutdown_smoke() {
714 each_ip(&mut |addr
| {
715 let a
= t
!(TcpListener
::bind(&addr
));
716 let _t
= thread
::spawn(move|| {
717 let mut c
= t
!(a
.accept()).0;
719 assert_eq
!(c
.read(&mut b
).unwrap(), 0);
723 let mut s
= t
!(TcpStream
::connect(&addr
));
724 t
!(s
.shutdown(Shutdown
::Write
));
725 assert
!(s
.write(&[1]).is_err());
727 assert_eq
!(t
!(s
.read(&mut b
)), 1);
733 fn close_readwrite_smoke() {
734 each_ip(&mut |addr
| {
735 let a
= t
!(TcpListener
::bind(&addr
));
736 let (tx
, rx
) = channel
::<()>();
737 let _t
= thread
::spawn(move|| {
738 let _s
= t
!(a
.accept());
743 let mut s
= t
!(TcpStream
::connect(&addr
));
744 let mut s2
= t
!(s
.try_clone());
746 // closing should prevent reads/writes
747 t
!(s
.shutdown(Shutdown
::Write
));
748 assert
!(s
.write(&[0]).is_err());
749 t
!(s
.shutdown(Shutdown
::Read
));
750 assert_eq
!(s
.read(&mut b
).unwrap(), 0);
752 // closing should affect previous handles
753 assert
!(s2
.write(&[0]).is_err());
754 assert_eq
!(s2
.read(&mut b
).unwrap(), 0);
756 // closing should affect new handles
757 let mut s3
= t
!(s
.try_clone());
758 assert
!(s3
.write(&[0]).is_err());
759 assert_eq
!(s3
.read(&mut b
).unwrap(), 0);
761 // make sure these don't die
762 let _
= s2
.shutdown(Shutdown
::Read
);
763 let _
= s2
.shutdown(Shutdown
::Write
);
764 let _
= s3
.shutdown(Shutdown
::Read
);
765 let _
= s3
.shutdown(Shutdown
::Write
);
771 fn close_read_wakes_up() {
772 each_ip(&mut |addr
| {
773 let a
= t
!(TcpListener
::bind(&addr
));
774 let (tx1
, rx
) = channel
::<()>();
775 let _t
= thread
::spawn(move|| {
776 let _s
= t
!(a
.accept());
780 let s
= t
!(TcpStream
::connect(&addr
));
781 let s2
= t
!(s
.try_clone());
782 let (tx
, rx
) = channel();
783 let _t
= thread
::spawn(move|| {
785 assert_eq
!(t
!(s2
.read(&mut [0])), 0);
786 tx
.send(()).unwrap();
788 // this should wake up the child thread
789 t
!(s
.shutdown(Shutdown
::Read
));
791 // this test will never finish if the child doesn't wake up
798 fn clone_while_reading() {
799 each_ip(&mut |addr
| {
800 let accept
= t
!(TcpListener
::bind(&addr
));
802 // Enqueue a thread to write to a socket
803 let (tx
, rx
) = channel();
804 let (txdone
, rxdone
) = channel();
805 let txdone2
= txdone
.clone();
806 let _t
= thread
::spawn(move|| {
807 let mut tcp
= t
!(TcpStream
::connect(&addr
));
810 txdone2
.send(()).unwrap();
813 // Spawn off a reading clone
814 let tcp
= t
!(accept
.accept()).0;
815 let tcp2
= t
!(tcp
.try_clone());
816 let txdone3
= txdone
.clone();
817 let _t
= thread
::spawn(move|| {
819 t
!(tcp2
.read(&mut [0]));
820 txdone3
.send(()).unwrap();
823 // Try to ensure that the reading clone is indeed reading
828 // clone the handle again while it's reading, then let it finish the
830 let _
= t
!(tcp
.try_clone());
831 tx
.send(()).unwrap();
832 rxdone
.recv().unwrap();
833 rxdone
.recv().unwrap();
838 fn clone_accept_smoke() {
839 each_ip(&mut |addr
| {
840 let a
= t
!(TcpListener
::bind(&addr
));
841 let a2
= t
!(a
.try_clone());
843 let _t
= thread
::spawn(move|| {
844 let _
= TcpStream
::connect(&addr
);
846 let _t
= thread
::spawn(move|| {
847 let _
= TcpStream
::connect(&addr
);
856 fn clone_accept_concurrent() {
857 each_ip(&mut |addr
| {
858 let a
= t
!(TcpListener
::bind(&addr
));
859 let a2
= t
!(a
.try_clone());
861 let (tx
, rx
) = channel();
862 let tx2
= tx
.clone();
864 let _t
= thread
::spawn(move|| {
865 tx
.send(t
!(a
.accept())).unwrap();
867 let _t
= thread
::spawn(move|| {
868 tx2
.send(t
!(a2
.accept())).unwrap();
871 let _t
= thread
::spawn(move|| {
872 let _
= TcpStream
::connect(&addr
);
874 let _t
= thread
::spawn(move|| {
875 let _
= TcpStream
::connect(&addr
);
885 let name
= if cfg
!(windows
) {"socket"}
else {"fd"}
;
886 let socket_addr
= next_test_ip4();
888 let listener
= t
!(TcpListener
::bind(&socket_addr
));
889 let listener_inner
= listener
.0.socket().as_inner();
890 let compare
= format
!("TcpListener {{ addr: {:?}, {}: {:?} }}",
891 socket_addr
, name
, listener_inner
);
892 assert_eq
!(format
!("{:?}", listener
), compare
);
894 let stream
= t
!(TcpStream
::connect(&("localhost",
895 socket_addr
.port())));
896 let stream_inner
= stream
.0.socket().as_inner();
897 let compare
= format
!("TcpStream {{ addr: {:?}, \
898 peer: {:?}, {}: {:?} }}",
899 stream
.local_addr().unwrap(),
900 stream
.peer_addr().unwrap(),
903 assert_eq
!(format
!("{:?}", stream
), compare
);
906 // FIXME: re-enabled bitrig/openbsd tests once their socket timeout code
907 // no longer has rounding errors.
908 #[cfg_attr(any(target_os = "bitrig", target_os = "netbsd", target_os = "openbsd"), ignore)]
911 let addr
= next_test_ip4();
912 let listener
= t
!(TcpListener
::bind(&addr
));
914 let stream
= t
!(TcpStream
::connect(&("localhost", addr
.port())));
915 let dur
= Duration
::new(15410, 0);
917 assert_eq
!(None
, t
!(stream
.read_timeout()));
919 t
!(stream
.set_read_timeout(Some(dur
)));
920 assert_eq
!(Some(dur
), t
!(stream
.read_timeout()));
922 assert_eq
!(None
, t
!(stream
.write_timeout()));
924 t
!(stream
.set_write_timeout(Some(dur
)));
925 assert_eq
!(Some(dur
), t
!(stream
.write_timeout()));
927 t
!(stream
.set_read_timeout(None
));
928 assert_eq
!(None
, t
!(stream
.read_timeout()));
930 t
!(stream
.set_write_timeout(None
));
931 assert_eq
!(None
, t
!(stream
.write_timeout()));
935 fn test_read_timeout() {
936 let addr
= next_test_ip4();
937 let listener
= t
!(TcpListener
::bind(&addr
));
939 let mut stream
= t
!(TcpStream
::connect(&("localhost", addr
.port())));
940 t
!(stream
.set_read_timeout(Some(Duration
::from_millis(1000))));
942 let mut buf
= [0; 10];
943 let wait
= Duration
::span(|| {
944 let kind
= stream
.read(&mut buf
).err().expect("expected error").kind();
945 assert
!(kind
== ErrorKind
::WouldBlock
|| kind
== ErrorKind
::TimedOut
);
947 assert
!(wait
> Duration
::from_millis(400));
951 fn test_read_with_timeout() {
952 let addr
= next_test_ip4();
953 let listener
= t
!(TcpListener
::bind(&addr
));
955 let mut stream
= t
!(TcpStream
::connect(&("localhost", addr
.port())));
956 t
!(stream
.set_read_timeout(Some(Duration
::from_millis(1000))));
958 let mut other_end
= t
!(listener
.accept()).0;
959 t
!(other_end
.write_all(b
"hello world"));
961 let mut buf
= [0; 11];
962 t
!(stream
.read(&mut buf
));
963 assert_eq
!(b
"hello world", &buf
[..]);
965 let wait
= Duration
::span(|| {
966 let kind
= stream
.read(&mut buf
).err().expect("expected error").kind();
967 assert
!(kind
== ErrorKind
::WouldBlock
|| kind
== ErrorKind
::TimedOut
);
969 assert
!(wait
> Duration
::from_millis(400));