1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
|
#![deny(warnings, rust_2018_idioms)]
use env_logger;
use futures::{Future, Stream};
use std::sync::mpsc::channel;
use std::{net, thread};
use tokio_tcp::{TcpListener, TcpStream};
macro_rules! t {
($e:expr) => {
match $e {
Ok(e) => e,
Err(e) => panic!("{} failed with {:?}", stringify!($e), e),
}
};
}
#[test]
fn connect() {
drop(env_logger::try_init());
let srv = t!(net::TcpListener::bind("127.0.0.1:0"));
let addr = t!(srv.local_addr());
let t = thread::spawn(move || t!(srv.accept()).0);
let stream = TcpStream::connect(&addr);
let mine = t!(stream.wait());
let theirs = t.join().unwrap();
assert_eq!(t!(mine.local_addr()), t!(theirs.peer_addr()));
assert_eq!(t!(theirs.local_addr()), t!(mine.peer_addr()));
}
#[test]
fn accept() {
drop(env_logger::try_init());
let srv = t!(TcpListener::bind(&t!("127.0.0.1:0".parse())));
let addr = t!(srv.local_addr());
let (tx, rx) = channel();
let client = srv
.incoming()
.map(move |t| {
tx.send(()).unwrap();
t
})
.into_future()
.map_err(|e| e.0);
assert!(rx.try_recv().is_err());
let t = thread::spawn(move || net::TcpStream::connect(&addr).unwrap());
let (mine, _remaining) = t!(client.wait());
let mine = mine.unwrap();
let theirs = t.join().unwrap();
assert_eq!(t!(mine.local_addr()), t!(theirs.peer_addr()));
assert_eq!(t!(theirs.local_addr()), t!(mine.peer_addr()));
}
#[test]
fn accept2() {
drop(env_logger::try_init());
let srv = t!(TcpListener::bind(&t!("127.0.0.1:0".parse())));
let addr = t!(srv.local_addr());
let t = thread::spawn(move || net::TcpStream::connect(&addr).unwrap());
let (tx, rx) = channel();
let client = srv
.incoming()
.map(move |t| {
tx.send(()).unwrap();
t
})
.into_future()
.map_err(|e| e.0);
assert!(rx.try_recv().is_err());
let (mine, _remaining) = t!(client.wait());
mine.unwrap();
t.join().unwrap();
}
#[cfg(target_os = "linux")]
mod linux {
use tokio_tcp::TcpStream;
use env_logger;
use futures::{future, Future};
use mio::unix::UnixReady;
use net2::TcpStreamExt;
use tokio_io::AsyncRead;
use std::io::Write;
use std::time::Duration;
use std::{net, thread};
#[test]
fn poll_hup() {
drop(env_logger::try_init());
let srv = t!(net::TcpListener::bind("127.0.0.1:0"));
let addr = t!(srv.local_addr());
let t = thread::spawn(move || {
let mut client = t!(srv.accept()).0;
client.set_linger(Some(Duration::from_millis(0))).unwrap();
client.write(b"hello world").unwrap();
thread::sleep(Duration::from_millis(200));
});
let mut stream = t!(TcpStream::connect(&addr).wait());
// Poll for HUP before reading.
future::poll_fn(|| stream.poll_read_ready(UnixReady::hup().into()))
.wait()
.unwrap();
// Same for write half
future::poll_fn(|| stream.poll_write_ready())
.wait()
.unwrap();
let mut buf = vec![0; 11];
// Read the data
future::poll_fn(|| stream.poll_read(&mut buf))
.wait()
.unwrap();
assert_eq!(b"hello world", &buf[..]);
t.join().unwrap();
}
}
|
