]> git.proxmox.com Git - rustc.git/blob - src/libstd/net/tcp.rs
Imported Upstream version 1.9.0+dfsg1
[rustc.git] / src / libstd / net / tcp.rs
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.
4 //
5 // Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
6 // http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
7 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
8 // option. This file may not be copied, modified, or distributed
9 // except according to those terms.
10
11 use prelude::v1::*;
12 use io::prelude::*;
13
14 use fmt;
15 use io;
16 use net::{ToSocketAddrs, SocketAddr, Shutdown};
17 use sys_common::net as net_imp;
18 use sys_common::{AsInner, FromInner, IntoInner};
19 use time::Duration;
20
21 /// A structure which represents a TCP stream between a local socket and a
22 /// remote socket.
23 ///
24 /// The socket will be closed when the value is dropped.
25 ///
26 /// # Examples
27 ///
28 /// ```no_run
29 /// use std::io::prelude::*;
30 /// use std::net::TcpStream;
31 ///
32 /// {
33 /// let mut stream = TcpStream::connect("127.0.0.1:34254").unwrap();
34 ///
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
39 /// ```
40 #[stable(feature = "rust1", since = "1.0.0")]
41 pub struct TcpStream(net_imp::TcpStream);
42
43 /// A structure representing a socket server.
44 ///
45 /// # Examples
46 ///
47 /// ```no_run
48 /// use std::net::{TcpListener, TcpStream};
49 /// use std::thread;
50 ///
51 /// let listener = TcpListener::bind("127.0.0.1:80").unwrap();
52 ///
53 /// fn handle_client(stream: TcpStream) {
54 /// // ...
55 /// }
56 ///
57 /// // accept connections and process them, spawning a new thread for each one
58 /// for stream in listener.incoming() {
59 /// match stream {
60 /// Ok(stream) => {
61 /// thread::spawn(move|| {
62 /// // connection succeeded
63 /// handle_client(stream)
64 /// });
65 /// }
66 /// Err(e) => { /* connection failed */ }
67 /// }
68 /// }
69 ///
70 /// // close the socket server
71 /// drop(listener);
72 /// ```
73 #[stable(feature = "rust1", since = "1.0.0")]
74 pub struct TcpListener(net_imp::TcpListener);
75
76 /// An infinite iterator over the connections from a `TcpListener`.
77 ///
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 }
82
83 impl TcpStream {
84 /// Opens a TCP connection to a remote host.
85 ///
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)
92 }
93
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> {
97 self.0.peer_addr()
98 }
99
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> {
103 self.0.socket_addr()
104 }
105
106 /// Shuts down the read, write, or both halves of this connection.
107 ///
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<()> {
113 self.0.shutdown(how)
114 }
115
116 /// Creates a new independently owned handle to the underlying socket.
117 ///
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
121 /// stream.
122 #[stable(feature = "rust1", since = "1.0.0")]
123 pub fn try_clone(&self) -> io::Result<TcpStream> {
124 self.0.duplicate().map(TcpStream)
125 }
126
127 /// Sets the read timeout to the timeout specified.
128 ///
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
131 /// method.
132 ///
133 /// # Note
134 ///
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)
141 }
142
143 /// Sets the write timeout to the timeout specified.
144 ///
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
147 /// method.
148 ///
149 /// # Note
150 ///
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)
157 }
158
159 /// Returns the read timeout of this socket.
160 ///
161 /// If the timeout is `None`, then `read` calls will block indefinitely.
162 ///
163 /// # Note
164 ///
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()
169 }
170
171 /// Returns the write timeout of this socket.
172 ///
173 /// If the timeout is `None`, then `write` calls will block indefinitely.
174 ///
175 /// # Note
176 ///
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()
181 }
182
183 /// Sets the value of the `TCP_NODELAY` option on this socket.
184 ///
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
189 /// small packets.
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)
193 }
194
195 /// Gets the value of the `TCP_NODELAY` option on this socket.
196 ///
197 /// For more information about this option, see [`set_nodelay`][link].
198 ///
199 /// [link]: #method.set_nodelay
200 #[stable(feature = "net2_mutators", since = "1.9.0")]
201 pub fn nodelay(&self) -> io::Result<bool> {
202 self.0.nodelay()
203 }
204
205 /// Sets the value for the `IP_TTL` option on this socket.
206 ///
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<()> {
211 self.0.set_ttl(ttl)
212 }
213
214 /// Gets the value of the `IP_TTL` option for this socket.
215 ///
216 /// For more information about this option, see [`set_ttl`][link].
217 ///
218 /// [link]: #method.set_ttl
219 #[stable(feature = "net2_mutators", since = "1.9.0")]
220 pub fn ttl(&self) -> io::Result<u32> {
221 self.0.ttl()
222 }
223
224 /// Get the value of the `SO_ERROR` option on this socket.
225 ///
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
228 /// calls.
229 #[stable(feature = "net2_mutators", since = "1.9.0")]
230 pub fn take_error(&self) -> io::Result<Option<io::Error>> {
231 self.0.take_error()
232 }
233
234 /// Moves this TCP stream into or out of nonblocking mode.
235 ///
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)
241 }
242 }
243
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)
249 }
250 }
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(()) }
255 }
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)
261 }
262 }
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(()) }
267 }
268
269 impl AsInner<net_imp::TcpStream> for TcpStream {
270 fn as_inner(&self) -> &net_imp::TcpStream { &self.0 }
271 }
272
273 impl FromInner<net_imp::TcpStream> for TcpStream {
274 fn from_inner(inner: net_imp::TcpStream) -> TcpStream { TcpStream(inner) }
275 }
276
277 impl IntoInner<net_imp::TcpStream> for TcpStream {
278 fn into_inner(self) -> net_imp::TcpStream { self.0 }
279 }
280
281 #[stable(feature = "rust1", since = "1.0.0")]
282 impl fmt::Debug for TcpStream {
283 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
284 self.0.fmt(f)
285 }
286 }
287
288 impl TcpListener {
289 /// Creates a new `TcpListener` which will be bound to the specified
290 /// address.
291 ///
292 /// The returned listener is ready for accepting connections.
293 ///
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.
297 ///
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)
303 }
304
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> {
308 self.0.socket_addr()
309 }
310
311 /// Creates a new independently owned handle to the underlying socket.
312 ///
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)
319 }
320
321 /// Accept a new incoming connection from this listener.
322 ///
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))
329 }
330
331 /// Returns an iterator over the connections being received on this
332 /// listener.
333 ///
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 }
339 }
340
341 /// Sets the value for the `IP_TTL` option on this socket.
342 ///
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<()> {
347 self.0.set_ttl(ttl)
348 }
349
350 /// Gets the value of the `IP_TTL` option for this socket.
351 ///
352 /// For more information about this option, see [`set_ttl`][link].
353 ///
354 /// [link]: #method.set_ttl
355 #[stable(feature = "net2_mutators", since = "1.9.0")]
356 pub fn ttl(&self) -> io::Result<u32> {
357 self.0.ttl()
358 }
359
360 /// Sets the value for the `IPV6_V6ONLY` option on this socket.
361 ///
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.
365 ///
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)
371 }
372
373 /// Gets the value of the `IPV6_V6ONLY` option for this socket.
374 ///
375 /// For more information about this option, see [`set_only_v6`][link].
376 ///
377 /// [link]: #method.set_only_v6
378 #[stable(feature = "net2_mutators", since = "1.9.0")]
379 pub fn only_v6(&self) -> io::Result<bool> {
380 self.0.only_v6()
381 }
382
383 /// Get the value of the `SO_ERROR` option on this socket.
384 ///
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
387 /// calls.
388 #[stable(feature = "net2_mutators", since = "1.9.0")]
389 pub fn take_error(&self) -> io::Result<Option<io::Error>> {
390 self.0.take_error()
391 }
392
393 /// Moves this TCP stream into or out of nonblocking mode.
394 ///
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)
400 }
401 }
402
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))
408 }
409 }
410
411 impl AsInner<net_imp::TcpListener> for TcpListener {
412 fn as_inner(&self) -> &net_imp::TcpListener { &self.0 }
413 }
414
415 impl FromInner<net_imp::TcpListener> for TcpListener {
416 fn from_inner(inner: net_imp::TcpListener) -> TcpListener {
417 TcpListener(inner)
418 }
419 }
420
421 impl IntoInner<net_imp::TcpListener> for TcpListener {
422 fn into_inner(self) -> net_imp::TcpListener { self.0 }
423 }
424
425 #[stable(feature = "rust1", since = "1.0.0")]
426 impl fmt::Debug for TcpListener {
427 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
428 self.0.fmt(f)
429 }
430 }
431
432 #[cfg(test)]
433 mod tests {
434 use prelude::v1::*;
435
436 use io::ErrorKind;
437 use io::prelude::*;
438 use net::*;
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};
443 use thread;
444
445 fn each_ip(f: &mut FnMut(SocketAddr)) {
446 f(next_test_ip4());
447 f(next_test_ip6());
448 }
449
450 macro_rules! t {
451 ($e:expr) => {
452 match $e {
453 Ok(t) => t,
454 Err(e) => panic!("received error for `{}`: {}", stringify!($e), e),
455 }
456 }
457 }
458
459 #[test]
460 fn bind_error() {
461 match TcpListener::bind("1.1.1.1:9999") {
462 Ok(..) => panic!(),
463 Err(e) =>
464 assert_eq!(e.kind(), ErrorKind::AddrNotAvailable),
465 }
466 }
467
468 #[test]
469 fn connect_error() {
470 match TcpStream::connect("0.0.0.0:1") {
471 Ok(..) => panic!(),
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()),
477 }
478 }
479
480 #[test]
481 fn listen_localhost() {
482 let socket_addr = next_test_ip4();
483 let listener = t!(TcpListener::bind(&socket_addr));
484
485 let _t = thread::spawn(move || {
486 let mut stream = t!(TcpStream::connect(&("localhost",
487 socket_addr.port())));
488 t!(stream.write(&[144]));
489 });
490
491 let mut stream = t!(listener.accept()).0;
492 let mut buf = [0];
493 t!(stream.read(&mut buf));
494 assert!(buf[0] == 144);
495 }
496
497 #[test]
498 fn connect_loopback() {
499 each_ip(&mut |addr| {
500 let acceptor = t!(TcpListener::bind(&addr));
501
502 let _t = thread::spawn(move|| {
503 let host = match addr {
504 SocketAddr::V4(..) => "127.0.0.1",
505 SocketAddr::V6(..) => "::1",
506 };
507 let mut stream = t!(TcpStream::connect(&(host, addr.port())));
508 t!(stream.write(&[66]));
509 });
510
511 let mut stream = t!(acceptor.accept()).0;
512 let mut buf = [0];
513 t!(stream.read(&mut buf));
514 assert!(buf[0] == 66);
515 })
516 }
517
518 #[test]
519 fn smoke_test() {
520 each_ip(&mut |addr| {
521 let acceptor = t!(TcpListener::bind(&addr));
522
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();
528 });
529
530 let (mut stream, addr) = t!(acceptor.accept());
531 let mut buf = [0];
532 t!(stream.read(&mut buf));
533 assert!(buf[0] == 99);
534 assert_eq!(addr, t!(rx.recv()));
535 })
536 }
537
538 #[test]
539 fn read_eof() {
540 each_ip(&mut |addr| {
541 let acceptor = t!(TcpListener::bind(&addr));
542
543 let _t = thread::spawn(move|| {
544 let _stream = t!(TcpStream::connect(&addr));
545 // Close
546 });
547
548 let mut stream = t!(acceptor.accept()).0;
549 let mut buf = [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);
554 })
555 }
556
557 #[test]
558 fn write_close() {
559 each_ip(&mut |addr| {
560 let acceptor = t!(TcpListener::bind(&addr));
561
562 let (tx, rx) = channel();
563 let _t = thread::spawn(move|| {
564 drop(t!(TcpStream::connect(&addr)));
565 tx.send(()).unwrap();
566 });
567
568 let mut stream = t!(acceptor.accept()).0;
569 rx.recv().unwrap();
570 let buf = [0];
571 match stream.write(&buf) {
572 Ok(..) => {}
573 Err(e) => {
574 assert!(e.kind() == ErrorKind::ConnectionReset ||
575 e.kind() == ErrorKind::BrokenPipe ||
576 e.kind() == ErrorKind::ConnectionAborted,
577 "unknown error: {}", e);
578 }
579 }
580 })
581 }
582
583 #[test]
584 fn multiple_connect_serial() {
585 each_ip(&mut |addr| {
586 let max = 10;
587 let acceptor = t!(TcpListener::bind(&addr));
588
589 let _t = thread::spawn(move|| {
590 for _ in 0..max {
591 let mut stream = t!(TcpStream::connect(&addr));
592 t!(stream.write(&[99]));
593 }
594 });
595
596 for stream in acceptor.incoming().take(max) {
597 let mut stream = t!(stream);
598 let mut buf = [0];
599 t!(stream.read(&mut buf));
600 assert_eq!(buf[0], 99);
601 }
602 })
603 }
604
605 #[test]
606 fn multiple_connect_interleaved_greedy_schedule() {
607 const MAX: usize = 10;
608 each_ip(&mut |addr| {
609 let acceptor = t!(TcpListener::bind(&addr));
610
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);
617 let mut buf = [0];
618 t!(stream.read(&mut buf));
619 assert!(buf[0] == i as u8);
620 });
621 }
622 });
623
624 connect(0, addr);
625 });
626
627 fn connect(i: usize, addr: SocketAddr) {
628 if i == MAX { return }
629
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]));
635 });
636 t.join().ok().unwrap();
637 }
638 }
639
640 #[test]
641 fn multiple_connect_interleaved_lazy_schedule() {
642 const MAX: usize = 10;
643 each_ip(&mut |addr| {
644 let acceptor = t!(TcpListener::bind(&addr));
645
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);
651 let mut buf = [0];
652 t!(stream.read(&mut buf));
653 assert!(buf[0] == 99);
654 });
655 }
656 });
657
658 connect(0, addr);
659 });
660
661 fn connect(i: usize, addr: SocketAddr) {
662 if i == MAX { return }
663
664 let t = thread::spawn(move|| {
665 let mut stream = t!(TcpStream::connect(&addr));
666 connect(i + 1, addr);
667 t!(stream.write(&[99]));
668 });
669 t.join().ok().unwrap();
670 }
671 }
672
673 #[test]
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());
681 });
682
683 let stream = t!(TcpStream::connect(&addr));
684 assert_eq!(addr, t!(stream.peer_addr()));
685 })
686 }
687
688 #[test]
689 fn partial_read() {
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();
696 let mut b = [0];
697 t!(cl.read(&mut b));
698 tx.send(()).unwrap();
699 });
700
701 let mut c = t!(TcpStream::connect(&addr));
702 let mut b = [0; 10];
703 assert_eq!(c.read(&mut b).unwrap(), 1);
704 t!(c.write(&[1]));
705 rx.recv().unwrap();
706 })
707 }
708
709 #[test]
710 fn double_bind() {
711 each_ip(&mut |addr| {
712 let _listener = t!(TcpListener::bind(&addr));
713 match TcpListener::bind(&addr) {
714 Ok(..) => panic!(),
715 Err(e) => {
716 assert!(e.kind() == ErrorKind::ConnectionRefused ||
717 e.kind() == ErrorKind::Other ||
718 e.kind() == ErrorKind::AddrInUse,
719 "unknown error: {} {:?}", e, e.kind());
720 }
721 }
722 })
723 }
724
725 #[test]
726 fn fast_rebind() {
727 each_ip(&mut |addr| {
728 let acceptor = t!(TcpListener::bind(&addr));
729
730 let _t = thread::spawn(move|| {
731 t!(TcpStream::connect(&addr));
732 });
733
734 t!(acceptor.accept());
735 drop(acceptor);
736 t!(TcpListener::bind(&addr));
737 });
738 }
739
740 #[test]
741 fn tcp_clone_smoke() {
742 each_ip(&mut |addr| {
743 let acceptor = t!(TcpListener::bind(&addr));
744
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);
750 t!(s.write(&[2]));
751 });
752
753 let mut s1 = t!(acceptor.accept()).0;
754 let s2 = t!(s1.try_clone());
755
756 let (tx1, rx1) = channel();
757 let (tx2, rx2) = channel();
758 let _t = thread::spawn(move|| {
759 let mut s2 = s2;
760 rx1.recv().unwrap();
761 t!(s2.write(&[1]));
762 tx2.send(()).unwrap();
763 });
764 tx1.send(()).unwrap();
765 let mut buf = [0, 0];
766 assert_eq!(s1.read(&mut buf).unwrap(), 1);
767 rx2.recv().unwrap();
768 })
769 }
770
771 #[test]
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();
777
778 let _t = thread::spawn(move|| {
779 let mut s = t!(TcpStream::connect(&addr));
780 t!(s.write(&[1]));
781 rx.recv().unwrap();
782 t!(s.write(&[2]));
783 rx.recv().unwrap();
784 });
785
786 let mut s1 = t!(acceptor.accept()).0;
787 let s2 = t!(s1.try_clone());
788
789 let (done, rx) = channel();
790 let _t = thread::spawn(move|| {
791 let mut s2 = s2;
792 let mut buf = [0, 0];
793 t!(s2.read(&mut buf));
794 tx2.send(()).unwrap();
795 done.send(()).unwrap();
796 });
797 let mut buf = [0, 0];
798 t!(s1.read(&mut buf));
799 tx1.send(()).unwrap();
800
801 rx.recv().unwrap();
802 })
803 }
804
805 #[test]
806 fn tcp_clone_two_write() {
807 each_ip(&mut |addr| {
808 let acceptor = t!(TcpListener::bind(&addr));
809
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));
815 });
816
817 let mut s1 = t!(acceptor.accept()).0;
818 let s2 = t!(s1.try_clone());
819
820 let (done, rx) = channel();
821 let _t = thread::spawn(move|| {
822 let mut s2 = s2;
823 t!(s2.write(&[1]));
824 done.send(()).unwrap();
825 });
826 t!(s1.write(&[2]));
827
828 rx.recv().unwrap();
829 })
830 }
831
832 #[test]
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;
838 let mut b = [0];
839 assert_eq!(c.read(&mut b).unwrap(), 0);
840 t!(c.write(&[1]));
841 });
842
843 let mut s = t!(TcpStream::connect(&addr));
844 t!(s.shutdown(Shutdown::Write));
845 assert!(s.write(&[1]).is_err());
846 let mut b = [0, 0];
847 assert_eq!(t!(s.read(&mut b)), 1);
848 assert_eq!(b[0], 1);
849 })
850 }
851
852 #[test]
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());
859 let _ = rx.recv();
860 });
861
862 let mut b = [0];
863 let mut s = t!(TcpStream::connect(&addr));
864 let mut s2 = t!(s.try_clone());
865
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);
871
872 // closing should affect previous handles
873 assert!(s2.write(&[0]).is_err());
874 assert_eq!(s2.read(&mut b).unwrap(), 0);
875
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);
880
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);
886 drop(tx);
887 })
888 }
889
890 #[test]
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());
897 let _ = rx.recv();
898 });
899
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|| {
904 let mut s2 = s2;
905 assert_eq!(t!(s2.read(&mut [0])), 0);
906 tx.send(()).unwrap();
907 });
908 // this should wake up the child thread
909 t!(s.shutdown(Shutdown::Read));
910
911 // this test will never finish if the child doesn't wake up
912 rx.recv().unwrap();
913 drop(tx1);
914 })
915 }
916
917 #[test]
918 fn clone_while_reading() {
919 each_ip(&mut |addr| {
920 let accept = t!(TcpListener::bind(&addr));
921
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));
928 rx.recv().unwrap();
929 t!(tcp.write(&[0]));
930 txdone2.send(()).unwrap();
931 });
932
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|| {
938 let mut tcp2 = tcp2;
939 t!(tcp2.read(&mut [0]));
940 txdone3.send(()).unwrap();
941 });
942
943 // Try to ensure that the reading clone is indeed reading
944 for _ in 0..50 {
945 thread::yield_now();
946 }
947
948 // clone the handle again while it's reading, then let it finish the
949 // read.
950 let _ = t!(tcp.try_clone());
951 tx.send(()).unwrap();
952 rxdone.recv().unwrap();
953 rxdone.recv().unwrap();
954 })
955 }
956
957 #[test]
958 fn clone_accept_smoke() {
959 each_ip(&mut |addr| {
960 let a = t!(TcpListener::bind(&addr));
961 let a2 = t!(a.try_clone());
962
963 let _t = thread::spawn(move|| {
964 let _ = TcpStream::connect(&addr);
965 });
966 let _t = thread::spawn(move|| {
967 let _ = TcpStream::connect(&addr);
968 });
969
970 t!(a.accept());
971 t!(a2.accept());
972 })
973 }
974
975 #[test]
976 fn clone_accept_concurrent() {
977 each_ip(&mut |addr| {
978 let a = t!(TcpListener::bind(&addr));
979 let a2 = t!(a.try_clone());
980
981 let (tx, rx) = channel();
982 let tx2 = tx.clone();
983
984 let _t = thread::spawn(move|| {
985 tx.send(t!(a.accept())).unwrap();
986 });
987 let _t = thread::spawn(move|| {
988 tx2.send(t!(a2.accept())).unwrap();
989 });
990
991 let _t = thread::spawn(move|| {
992 let _ = TcpStream::connect(&addr);
993 });
994 let _t = thread::spawn(move|| {
995 let _ = TcpStream::connect(&addr);
996 });
997
998 rx.recv().unwrap();
999 rx.recv().unwrap();
1000 })
1001 }
1002
1003 #[test]
1004 fn debug() {
1005 let name = if cfg!(windows) {"socket"} else {"fd"};
1006 let socket_addr = next_test_ip4();
1007
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);
1013
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(),
1021 name,
1022 stream_inner);
1023 assert_eq!(format!("{:?}", stream), compare);
1024 }
1025
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)]
1029 #[test]
1030 fn timeouts() {
1031 let addr = next_test_ip4();
1032 let listener = t!(TcpListener::bind(&addr));
1033
1034 let stream = t!(TcpStream::connect(&("localhost", addr.port())));
1035 let dur = Duration::new(15410, 0);
1036
1037 assert_eq!(None, t!(stream.read_timeout()));
1038
1039 t!(stream.set_read_timeout(Some(dur)));
1040 assert_eq!(Some(dur), t!(stream.read_timeout()));
1041
1042 assert_eq!(None, t!(stream.write_timeout()));
1043
1044 t!(stream.set_write_timeout(Some(dur)));
1045 assert_eq!(Some(dur), t!(stream.write_timeout()));
1046
1047 t!(stream.set_read_timeout(None));
1048 assert_eq!(None, t!(stream.read_timeout()));
1049
1050 t!(stream.set_write_timeout(None));
1051 assert_eq!(None, t!(stream.write_timeout()));
1052 drop(listener);
1053 }
1054
1055 #[test]
1056 fn test_read_timeout() {
1057 let addr = next_test_ip4();
1058 let listener = t!(TcpListener::bind(&addr));
1059
1060 let mut stream = t!(TcpStream::connect(&("localhost", addr.port())));
1061 t!(stream.set_read_timeout(Some(Duration::from_millis(1000))));
1062
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));
1068 drop(listener);
1069 }
1070
1071 #[test]
1072 fn test_read_with_timeout() {
1073 let addr = next_test_ip4();
1074 let listener = t!(TcpListener::bind(&addr));
1075
1076 let mut stream = t!(TcpStream::connect(&("localhost", addr.port())));
1077 t!(stream.set_read_timeout(Some(Duration::from_millis(1000))));
1078
1079 let mut other_end = t!(listener.accept()).0;
1080 t!(other_end.write_all(b"hello world"));
1081
1082 let mut buf = [0; 11];
1083 t!(stream.read(&mut buf));
1084 assert_eq!(b"hello world", &buf[..]);
1085
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));
1090 drop(listener);
1091 }
1092
1093 #[test]
1094 fn nodelay() {
1095 let addr = next_test_ip4();
1096 let _listener = t!(TcpListener::bind(&addr));
1097
1098 let stream = t!(TcpStream::connect(&("localhost", addr.port())));
1099
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()));
1105 }
1106
1107 #[test]
1108 fn ttl() {
1109 let ttl = 100;
1110
1111 let addr = next_test_ip4();
1112 let listener = t!(TcpListener::bind(&addr));
1113
1114 t!(listener.set_ttl(ttl));
1115 assert_eq!(ttl, t!(listener.ttl()));
1116
1117 let stream = t!(TcpStream::connect(&("localhost", addr.port())));
1118
1119 t!(stream.set_ttl(ttl));
1120 assert_eq!(ttl, t!(stream.ttl()));
1121 }
1122
1123 #[test]
1124 fn set_nonblocking() {
1125 let addr = next_test_ip4();
1126 let listener = t!(TcpListener::bind(&addr));
1127
1128 t!(listener.set_nonblocking(true));
1129 t!(listener.set_nonblocking(false));
1130
1131 let mut stream = t!(TcpStream::connect(&("localhost", addr.port())));
1132
1133 t!(stream.set_nonblocking(false));
1134 t!(stream.set_nonblocking(true));
1135
1136 let mut buf = [0];
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),
1141 }
1142 }
1143 }