tokio 0.1.3

An event-driven, non-blocking I/O platform for writing asynchronous I/O backed applications.
Documentation 
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

extern crate env_logger;
extern crate tokio;
extern crate mio;
extern crate futures;

use std::{net, thread};
use std::sync::mpsc::channel;

use tokio::net::{TcpListener, TcpStream};
use tokio::prelude::*;


macro_rules! t {
    ($e:expr) => (match $e {
        Ok(e) => e,
        Err(e) => panic!("{} failed with {:?}", stringify!($e), e),
    })
}

#[test]
fn connect() {
    drop(env_logger::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::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::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(unix)]
mod unix {
    use tokio::net::TcpStream;
    use tokio::prelude::*;

    use env_logger;
    use futures::future;
    use mio::unix::UnixReady;

    use std::{net, thread};
    use std::time::Duration;

    #[test]
    fn poll_hup() {
        drop(env_logger::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.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();
    }
}