This repository has been archived by the owner on Jul 13, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 34
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: Add a
/status handler on the ws port
This adds some logic to parse an incoming HTTP request, peek at the headers, and then route the response appropriately.
- Loading branch information
1 parent
843dc68
commit 2430d35
Showing
9 changed files
with
231 additions
and
19 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -25,3 +25,5 @@ pypy/ | |
| src/ | ||
| .tox/ | ||
| .eggs/ | ||
| autopush_rs/target | ||
| autopush_rs/_native* | ||
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,86 @@ | ||
| //! A future to figure out where we're going to dispatch a TCP socket. | ||
| //! | ||
| //! When the websocket server receives a TCP connection it may be a websocket | ||
| //! request or a general HTTP request. Right now the websocket library we're | ||
| //! using, Tungstenite, doesn't have built-in support for handling this | ||
| //! situation, so we roll our own. | ||
| //! | ||
| //! The general idea here is that we're going to read just enough data off the | ||
| //! socket to parse an initial HTTP request. This request will be parsed by the | ||
| //! `httparse` crate. Once we've got a request we take a look at the headers and | ||
| //! if we find a websocket upgrade we classify it as a websocket request. If | ||
| //! it's otherwise a `/status` request, we return that we're supposed to get the | ||
| //! status, and finally after all that if it doesn't match we return an error. | ||
| //! | ||
| //! This is basically a "poor man's" HTTP router and while it should be good | ||
| //! enough for now it should probably be extended/refactored in the future! | ||
| //! | ||
| //! Note that also to implement this we buffer the request that we read in | ||
| //! memory and then attach that to a socket once we've classified what kind of | ||
| //! socket this is. That's done to replay the bytes we read again for the | ||
| //! tungstenite library, which'll duplicate header parsing but we don't have | ||
| //! many other options for now! | ||
|
|
||
| use bytes::BytesMut; | ||
| use futures::{Future, Poll}; | ||
| use httparse; | ||
| use tokio_core::net::TcpStream; | ||
| use tokio_io::AsyncRead; | ||
|
|
||
| use errors::*; | ||
| use server::webpush_io::WebpushIo; | ||
|
|
||
| pub struct Dispatch { | ||
| socket: Option<TcpStream>, | ||
| data: BytesMut, | ||
| } | ||
|
|
||
| pub enum RequestType { | ||
| Websocket, | ||
| Status, | ||
| } | ||
|
|
||
| impl Dispatch { | ||
| pub fn new(socket: TcpStream) -> Dispatch { | ||
| Dispatch { | ||
| socket: Some(socket), | ||
| data: BytesMut::new(), | ||
| } | ||
| } | ||
| } | ||
|
|
||
| impl Future for Dispatch { | ||
| type Item = (WebpushIo, RequestType); | ||
| type Error = Error; | ||
|
|
||
| fn poll(&mut self) -> Poll<(WebpushIo, RequestType), Error> { | ||
| loop { | ||
| if self.data.len() == self.data.capacity() { | ||
| self.data.reserve(16); // get some extra space | ||
| } | ||
| if try_ready!(self.socket.as_mut().unwrap().read_buf(&mut self.data)) == 0 { | ||
| return Err("early eof".into()) | ||
| } | ||
| let ty = { | ||
| let mut headers = [httparse::EMPTY_HEADER; 16]; | ||
| let mut req = httparse::Request::new(&mut headers); | ||
| match req.parse(&self.data)? { | ||
| httparse::Status::Complete(_) => {} | ||
| httparse::Status::Partial => continue, | ||
| } | ||
|
|
||
| if req.headers.iter().any(|h| h.name == "Upgrade") { | ||
| RequestType::Websocket | ||
| } else if req.path == Some("/status") { | ||
| RequestType::Status | ||
| } else { | ||
| debug!("unknown http request {:?}", req); | ||
| return Err("unknown http request".into()) | ||
| } | ||
| }; | ||
|
|
||
| let tcp = self.socket.take().unwrap(); | ||
| return Ok((WebpushIo::new(tcp, self.data.take()), ty).into()) | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| //! I/O wrapper created through `Dispatch` | ||
| //! | ||
| //! Most I/O happens through just raw TCP sockets, but at the beginning of a | ||
| //! request we'll take a look at the headers and figure out where to route it. | ||
| //! After that, for tungstenite the websocket library, we'll want to replay the | ||
| //! data we already read as there's no ability to pass this in currently. That | ||
| //! means we'll parse headers twice, but alas! | ||
|
|
||
| use std::io::{self, Read, Write}; | ||
|
|
||
| use bytes::BytesMut; | ||
| use futures::Poll; | ||
| use tokio_core::net::TcpStream; | ||
| use tokio_io::{AsyncRead, AsyncWrite}; | ||
|
|
||
| pub struct WebpushIo { | ||
| tcp: TcpStream, | ||
| header_to_read: Option<BytesMut>, | ||
| } | ||
|
|
||
| impl WebpushIo { | ||
| pub fn new(tcp: TcpStream, header: BytesMut) -> WebpushIo { | ||
| WebpushIo { | ||
| tcp: tcp, | ||
| header_to_read: Some(header), | ||
| } | ||
| } | ||
| } | ||
|
|
||
| impl Read for WebpushIo { | ||
| fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> { | ||
| // Start off by replaying the bytes already read, and after that just | ||
| // delegate everything to the internal `TcpStream` | ||
| if let Some(ref mut header) = self.header_to_read { | ||
| let n = (&header[..]).read(buf)?; | ||
| header.split_to(n); | ||
| if buf.len() == 0 || n > 0 { | ||
| return Ok(n) | ||
| } | ||
| } | ||
| self.header_to_read = None; | ||
| self.tcp.read(buf) | ||
| } | ||
| } | ||
|
|
||
| // All `write` calls are routed through the `TcpStream` instance directly as we | ||
| // don't buffer this at all. | ||
| impl Write for WebpushIo { | ||
| fn write(&mut self, buf: &[u8]) -> io::Result<usize> { | ||
| self.tcp.write(buf) | ||
| } | ||
|
|
||
| fn flush(&mut self) -> io::Result<()> { | ||
| self.tcp.flush() | ||
| } | ||
| } | ||
|
|
||
| impl AsyncRead for WebpushIo { | ||
| } | ||
|
|
||
| impl AsyncWrite for WebpushIo { | ||
| fn shutdown(&mut self) -> Poll<(), io::Error> { | ||
| AsyncWrite::shutdown(&mut self.tcp) | ||
| } | ||
| } |