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
|
use mio::{Ready,Token,Event};
use mio::tcp::TcpStream;
use std::net::SocketAddr;
use std::io::ErrorKind;
use eventloop::{Machine,Context};
use netbuf::Buf;
// TODO: timeouts
// TODO: Set a max size on the buffers and provide backpressure
pub struct ItfHttp {
sock: TcpStream,
addr: SocketAddr,
io: Token,
ioreg: Ready,
rbuf: Buf,
wbuf: Buf
}
impl ItfHttp {
pub fn new(sock: TcpStream, addr: SocketAddr) -> ItfHttp {
ItfHttp {
sock: sock,
addr: addr,
io: Token(0),
ioreg: Ready::readable(),
rbuf: Buf::new(),
wbuf: Buf::new()
}
}
fn handle_read(&mut self, ctx: &mut Context) {
match self.rbuf.read_from(&mut self.sock) {
Err(err) => {
match err.kind() {
ErrorKind::WouldBlock | ErrorKind::Interrupted => { },
_ => {
debug!("{}: Read error: {}", self.addr, err);
self.remove(ctx);
}
}
return;
},
Ok(0) => { // This behaviour isn't documented, unsure if it's intended.
debug!("{}: Connection closed", self.addr);
self.remove(ctx);
return;
},
_ => {}
}
// This is where we parse the stuff, generate a request object, throw the request object
// through some handlers to get a response and send it back. But for now let's just act
// like we wanted to be an echo server all along.
self.rbuf.write_to(&mut self.wbuf).unwrap();
}
fn handle_write(&mut self, ctx: &mut Context) {
if let Err(err) = self.wbuf.write_to(&mut self.sock) {
match err.kind() {
ErrorKind::WouldBlock | ErrorKind::Interrupted => { },
_ => {
debug!("{}: Write error: {}", self.addr, err);
self.remove(ctx);
}
}
}
}
fn set_ioreg(&mut self, ctx: &mut Context) {
// Optimization: If the write buffer was empty before but contains data now, we can already
// try to write it out at this point. This way we can avoid going through the event loop if
// the OS buffers have enough space.
// TODO: Measure how effective this is in practice; if this only tends to happen on the
// first write to a new socket then it might not be worth the effort.
if self.wbuf.len() > 0 && !self.ioreg.contains(Ready::writable()) {
self.handle_write(ctx);
if ctx.is_removed() {
return;
}
}
let mut reg = Ready::readable();
if self.wbuf.len() > 0 {
reg.insert(Ready::writable());
}
if reg != self.ioreg {
self.ioreg = reg;
ctx.reregister(&self.sock, self.io, self.ioreg);
}
}
fn remove(&mut self, ctx: &mut Context) {
ctx.deregister(&self.sock);
ctx.unassign(self.io);
ctx.remove();
}
}
impl Machine for ItfHttp {
fn init(&mut self, ctx: &mut Context) {
self.io = ctx.assign();
ctx.register(&self.sock, self.io, self.ioreg);
}
fn handle(&mut self, ctx: &mut Context, ev: Event) {
if ev.kind().is_readable() {
self.handle_read(ctx);
}
if !ctx.is_removed() && ev.kind().is_writable() {
self.handle_write(ctx);
}
if !ctx.is_removed() {
self.set_ioreg(ctx);
}
}
}
|
