swindon 0.7.8

An HTTP edge (frontend) server with smart websockets support 
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

use std::sync::Arc;

use futures::{Async, Future};
use futures::stream::{Stream};
use tk_http::Status;
use tk_http::server::{Error, Codec, RecvMode};
use tk_http::server as http;
use tk_http::websocket::{ServerCodec as WebsocketCodec, Accept};
use tk_bufstream::{ReadBuf, WriteBuf};
use futures::future::{ok};
use tokio_core::reactor::Handle;
use tokio_io::{AsyncRead, AsyncWrite};

use config::Config;
use incoming::{Request, Input, Debug, Reply, Encoder};
use default_error_page::serve_error_page;


struct WebsockReply {
    rdata: Option<(Arc<Config>, Debug, Accept)>,
    handle: Handle,
}


impl<S: AsyncRead + AsyncWrite + 'static> Codec<S> for WebsockReply {
    type ResponseFuture = Reply<S>;
    fn recv_mode(&mut self) -> RecvMode {
        RecvMode::hijack()
    }
    fn data_received(&mut self, data: &[u8], end: bool)
        -> Result<Async<usize>, Error>
    {
        assert!(end);
        assert!(data.len() == 0);
        Ok(Async::Ready(0))
    }
    fn start_response(&mut self, e: http::Encoder<S>) -> Reply<S> {
        let (config, debug, accept) = self.rdata.take()
            .expect("start response called once");
        let mut e = Encoder::new(e, (config, debug));
        e.status(Status::SwitchingProtocol);
        e.add_header("Connection", "upgrade");
        e.add_header("Upgrade", "websocket");
        e.format_header("Sec-Websocket-Accept", &accept);
        e.done_headers();
        Box::new(ok(e.done()))
    }
    fn hijack(&mut self, write_buf: WriteBuf<S>, read_buf: ReadBuf<S>) {
        let inp = read_buf.framed(WebsocketCodec);
        let out = write_buf.framed(WebsocketCodec);
        // TODO(tailhook) convert Ping to Pong (and Close ?) before echoing
        self.handle.spawn(inp.forward(out)
            .map(|_| ())
            // TODO(tailhook) check error reporting
            .map_err(|e| info!("Websocket error: {}", e)))
    }
}

pub fn serve<S: AsyncRead + AsyncWrite + 'static>(inp: Input) -> Request<S> {
    match inp.headers.get_websocket_upgrade() {
        Ok(Some(ws)) => {
            Box::new(WebsockReply {
                rdata: Some((inp.config.clone(), inp.debug, ws.accept)),
                handle: inp.handle.clone(),
            })
        }
        Ok(None) => {
            serve_error_page(Status::NotFound, inp)
        }
        Err(()) => {
            serve_error_page(Status::BadRequest, inp)
        }
    }
}