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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
|
// You can run this example from the root of the mio repo:
// cargo run --example tcp_server --features="os-poll net"
use mio::event::Event;
use mio::net::{TcpListener, TcpStream};
use mio::{Events, Interest, Poll, Registry, Token};
use std::collections::HashMap;
use std::io::{self, Read, Write};
use std::str::from_utf8;
// Setup some tokens to allow us to identify which event is for which socket.
const SERVER: Token = Token(0);
// Some data we'll send over the connection.
const DATA: &[u8] = b"Hello world!\n";
fn main() -> io::Result<()> {
env_logger::init();
// Create a poll instance.
let mut poll = Poll::new()?;
// Create storage for events.
let mut events = Events::with_capacity(128);
// Setup the TCP server socket.
let addr = "127.0.0.1:9000".parse().unwrap();
let mut server = TcpListener::bind(addr)?;
// Register the server with poll we can receive events for it.
poll.registry()
.register(&mut server, SERVER, Interest::READABLE)?;
// Map of `Token` -> `TcpStream`.
let mut connections = HashMap::new();
// Unique token for each incoming connection.
let mut unique_token = Token(SERVER.0 + 1);
println!("You can connect to the server using `nc`:");
println!(" $ nc 127.0.0.1 9000");
println!("You'll see our welcome message and anything you type will be printed here.");
loop {
poll.poll(&mut events, None)?;
for event in events.iter() {
match event.token() {
SERVER => loop {
// Received an event for the TCP server socket, which
// indicates we can accept an connection.
let (mut connection, address) = match server.accept() {
Ok((connection, address)) => (connection, address),
Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
// If we get a `WouldBlock` error we know our
// listener has no more incoming connections queued,
// so we can return to polling and wait for some
// more.
break;
}
Err(e) => {
// If it was any other kind of error, something went
// wrong and we terminate with an error.
return Err(e);
}
};
println!("Accepted connection from: {}", address);
let token = next(&mut unique_token);
poll.registry().register(
&mut connection,
token,
Interest::READABLE.add(Interest::WRITABLE),
)?;
connections.insert(token, connection);
},
token => {
// Maybe received an event for a TCP connection.
let done = if let Some(connection) = connections.get_mut(&token) {
handle_connection_event(poll.registry(), connection, event)?
} else {
// Sporadic events happen, we can safely ignore them.
false
};
if done {
if let Some(mut connection) = connections.remove(&token) {
poll.registry().deregister(&mut connection)?;
}
}
}
}
}
}
}
fn next(current: &mut Token) -> Token {
let next = current.0;
current.0 += 1;
Token(next)
}
/// Returns `true` if the connection is done.
fn handle_connection_event(
registry: &Registry,
connection: &mut TcpStream,
event: &Event,
) -> io::Result<bool> {
if event.is_writable() {
// We can (maybe) write to the connection.
match connection.write(DATA) {
// We want to write the entire `DATA` buffer in a single go. If we
// write less we'll return a short write error (same as
// `io::Write::write_all` does).
Ok(n) if n < DATA.len() => return Err(io::ErrorKind::WriteZero.into()),
Ok(_) => {
// After we've written something we'll reregister the connection
// to only respond to readable events.
registry.reregister(connection, event.token(), Interest::READABLE)?
}
// Would block "errors" are the OS's way of saying that the
// connection is not actually ready to perform this I/O operation.
Err(ref err) if would_block(err) => {}
// Got interrupted (how rude!), we'll try again.
Err(ref err) if interrupted(err) => {
return handle_connection_event(registry, connection, event)
}
// Other errors we'll consider fatal.
Err(err) => return Err(err),
}
}
if event.is_readable() {
let mut connection_closed = false;
let mut received_data = vec![0; 4096];
let mut bytes_read = 0;
// We can (maybe) read from the connection.
loop {
match connection.read(&mut received_data[bytes_read..]) {
Ok(0) => {
// Reading 0 bytes means the other side has closed the
// connection or is done writing, then so are we.
connection_closed = true;
break;
}
Ok(n) => {
bytes_read += n;
if bytes_read == received_data.len() {
received_data.resize(received_data.len() + 1024, 0);
}
}
// Would block "errors" are the OS's way of saying that the
// connection is not actually ready to perform this I/O operation.
Err(ref err) if would_block(err) => break,
Err(ref err) if interrupted(err) => continue,
// Other errors we'll consider fatal.
Err(err) => return Err(err),
}
}
if bytes_read != 0 {
let received_data = &received_data[..bytes_read];
if let Ok(str_buf) = from_utf8(received_data) {
println!("Received data: {}", str_buf.trim_end());
} else {
println!("Received (none UTF-8) data: {:?}", received_data);
}
}
if connection_closed {
println!("Connection closed");
return Ok(true);
}
}
Ok(false)
}
fn would_block(err: &io::Error) -> bool {
err.kind() == io::ErrorKind::WouldBlock
}
fn interrupted(err: &io::Error) -> bool {
err.kind() == io::ErrorKind::Interrupted
}
|
