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
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
//! Send HTTP requests and responses asynchronously.
//!
//! This module has both an `HttpClientCodec` for an async HTTP client and an
//! `HttpServerCodec` for an async HTTP server.
use bytes::BufMut;
use bytes::BytesMut;
use hyper;
use hyper::buffer::BufReader;
use hyper::http::h1::parse_request;
use hyper::http::h1::parse_response;
use hyper::http::h1::Incoming;
use hyper::http::RawStatus;
use hyper::method::Method;
use hyper::status::StatusCode;
use hyper::uri::RequestUri;
use std::error::Error;
use std::fmt::{self, Display, Formatter};
use std::io::{self, Write};
use tokio_io::codec::{Decoder, Encoder};

#[derive(Copy, Clone, Debug)]
///A codec to be used with `tokio` codecs that can serialize HTTP requests and
///deserialize HTTP responses. One can use this on it's own without websockets to
///make a very bare async HTTP server.
///
///# Example
///```rust,no_run
///# extern crate tokio_core;
///# extern crate tokio_io;
///# extern crate websocket;
///# extern crate hyper;
///use websocket::async::HttpClientCodec;
///# use websocket::async::futures::{Future, Sink, Stream};
///# use tokio_core::net::TcpStream;
///# use tokio_core::reactor::Core;
///# use tokio_io::AsyncRead;
///# use hyper::http::h1::Incoming;
///# use hyper::version::HttpVersion;
///# use hyper::header::Headers;
///# use hyper::method::Method;
///# use hyper::uri::RequestUri;
///
///# fn main() {
///let mut core = Core::new().unwrap();
///let addr = "crouton.net".parse().unwrap();
///
///let f = TcpStream::connect(&addr, &core.handle())
///    .and_then(|s| {
///        Ok(s.framed(HttpClientCodec))
///    })
///    .and_then(|s| {
///        s.send(Incoming {
///            version: HttpVersion::Http11,
///            subject: (Method::Get, RequestUri::AbsolutePath("/".to_string())),
///            headers: Headers::new(),
///        })
///    })
///    .map_err(|e| e.into())
///    .and_then(|s| s.into_future().map_err(|(e, _)| e))
///    .map(|(m, _)| println!("You got a crouton: {:?}", m));
///
///core.run(f).unwrap();
///# }
///```
pub struct HttpClientCodec;

fn split_off_http(src: &mut BytesMut) -> Option<BytesMut> {
	match src.windows(4).position(|i| i == b"\r\n\r\n") {
		Some(p) => Some(src.split_to(p + 4)),
		None => None,
	}
}

impl Encoder for HttpClientCodec {
	type Item = Incoming<(Method, RequestUri)>;
	type Error = io::Error;

	fn encode(&mut self, item: Self::Item, dst: &mut BytesMut) -> Result<(), Self::Error> {
		// TODO: optomize this!
		let request = format!(
			"{} {} {}\r\n{}\r\n",
			item.subject.0, item.subject.1, item.version, item.headers
		);
		let byte_len = request.as_bytes().len();
		if byte_len > dst.remaining_mut() {
			dst.reserve(byte_len);
		}
		dst.writer().write(request.as_bytes()).map(|_| ())
	}
}

impl Decoder for HttpClientCodec {
	type Item = Incoming<RawStatus>;
	type Error = HttpCodecError;

	fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
		// check if we get a request from hyper
		// TODO: this is ineffecient, but hyper does not give us a better way to parse
		match split_off_http(src) {
			Some(buf) => {
				let mut reader = BufReader::with_capacity(&*buf as &[u8], buf.len());
				let res = match parse_response(&mut reader) {
					Err(hyper::Error::Io(ref e)) if e.kind() == io::ErrorKind::UnexpectedEof => {
						return Ok(None)
					}
					Err(hyper::Error::TooLarge) => return Ok(None),
					Err(e) => return Err(e.into()),
					Ok(r) => r,
				};
				Ok(Some(res))
			}
			None => Ok(None),
		}
	}
}

///A codec that can be used with streams implementing `AsyncRead + AsyncWrite`
///that can serialize HTTP responses and deserialize HTTP requests. Using this
///with an async `TcpStream` will give you a very bare async HTTP server.
///
///This crate sends out one HTTP request / response in order to perform the websocket
///handshake then never talks HTTP again. Because of this an async HTTP implementation
///is needed.
///
///# Example
///
///```rust,no_run
///# extern crate tokio_core;
///# extern crate tokio_io;
///# extern crate websocket;
///# extern crate hyper;
///# use std::io;
///use websocket::async::HttpServerCodec;
///# use websocket::async::futures::{Future, Sink, Stream};
///# use tokio_core::net::TcpStream;
///# use tokio_core::reactor::Core;
///# use tokio_io::AsyncRead;
///# use hyper::http::h1::Incoming;
///# use hyper::version::HttpVersion;
///# use hyper::header::Headers;
///# use hyper::method::Method;
///# use hyper::uri::RequestUri;
///# use hyper::status::StatusCode;
///# fn main() {
///
///let mut core = Core::new().unwrap();
///let addr = "nothing-to-see-here.com".parse().unwrap();
///
///let f = TcpStream::connect(&addr, &core.handle())
///   .map(|s| s.framed(HttpServerCodec))
///   .map_err(|e| e.into())
///   .and_then(|s| s.into_future().map_err(|(e, _)| e))
///   .and_then(|(m, s)| match m {
///       Some(ref m) if m.subject.0 == Method::Get => Ok(s),
///       _ => panic!(),
///   })
///   .and_then(|stream| {
///       stream
///          .send(Incoming {
///               version: HttpVersion::Http11,
///               subject: StatusCode::NotFound,
///               headers: Headers::new(),
///           })
///           .map_err(|e| e.into())
///   });
///
///core.run(f).unwrap();
///# }
///```
#[derive(Copy, Clone, Debug)]
pub struct HttpServerCodec;

impl Encoder for HttpServerCodec {
	type Item = Incoming<StatusCode>;
	type Error = io::Error;

	fn encode(&mut self, item: Self::Item, dst: &mut BytesMut) -> Result<(), Self::Error> {
		// TODO: optomize this!
		let response = format!("{} {}\r\n{}\r\n", item.version, item.subject, item.headers);
		let byte_len = response.as_bytes().len();
		if byte_len > dst.remaining_mut() {
			dst.reserve(byte_len);
		}
		dst.writer().write(response.as_bytes()).map(|_| ())
	}
}

impl Decoder for HttpServerCodec {
	type Item = Incoming<(Method, RequestUri)>;
	type Error = HttpCodecError;

	fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
		// check if we get a request from hyper
		// TODO: this is ineffecient, but hyper does not give us a better way to parse
		match split_off_http(src) {
			Some(buf) => {
				let mut reader = BufReader::with_capacity(&*buf as &[u8], buf.len());
				let res = match parse_request(&mut reader) {
					Err(hyper::Error::Io(ref e)) if e.kind() == io::ErrorKind::UnexpectedEof => {
						return Ok(None);
					}
					Err(hyper::Error::TooLarge) => return Ok(None),
					Err(e) => return Err(e.into()),
					Ok(r) => r,
				};
				Ok(Some(res))
			}
			None => Ok(None),
		}
	}
}

/// Any error that can happen during the writing or parsing of HTTP requests
/// and responses. This consists of HTTP parsing errors (the `Http` variant) and
/// errors that can occur when writing to IO (the `Io` variant).
#[derive(Debug)]
pub enum HttpCodecError {
	/// An error that occurs during the writing or reading of HTTP data
	/// from a socket.
	Io(io::Error),
	/// An error that occurs during the parsing of an HTTP request or response.
	Http(hyper::Error),
}

impl Display for HttpCodecError {
	fn fmt(&self, fmt: &mut Formatter) -> Result<(), fmt::Error> {
		fmt.write_str(self.description())
	}
}

impl Error for HttpCodecError {
	fn description(&self) -> &str {
		match *self {
			HttpCodecError::Io(ref e) => e.description(),
			HttpCodecError::Http(ref e) => e.description(),
		}
	}

	fn cause(&self) -> Option<&Error> {
		match *self {
			HttpCodecError::Io(ref error) => Some(error),
			HttpCodecError::Http(ref error) => Some(error),
		}
	}
}

impl From<io::Error> for HttpCodecError {
	fn from(err: io::Error) -> HttpCodecError {
		HttpCodecError::Io(err)
	}
}

impl From<hyper::Error> for HttpCodecError {
	fn from(err: hyper::Error) -> HttpCodecError {
		HttpCodecError::Http(err)
	}
}

#[cfg(test)]
mod tests {
	use super::*;
	use futures::{Future, Sink, Stream};
	use hyper::header::Headers;
	use hyper::version::HttpVersion;
	use std::io::Cursor;
	use stream::ReadWritePair;
	use tokio_core::reactor::Core;
	use tokio_io::AsyncRead;

	#[test]
	fn test_client_http_codec() {
		let mut core = Core::new().unwrap();
		let response = "HTTP/1.1 404 Not Found\r\n\r\npssst extra data here";
		let input = Cursor::new(response.as_bytes());
		let output = Cursor::new(Vec::new());

		let f = ReadWritePair(input, output)
			.framed(HttpClientCodec)
			.send(Incoming {
				version: HttpVersion::Http11,
				subject: (Method::Get, RequestUri::AbsolutePath("/".to_string())),
				headers: Headers::new(),
			})
			.map_err(|e| e.into())
			.and_then(|s| s.into_future().map_err(|(e, _)| e))
			.and_then(|(m, _)| match m {
				Some(ref m) if StatusCode::from_u16(m.subject.0) == StatusCode::NotFound => Ok(()),
				_ => Err(io::Error::new(io::ErrorKind::Other, "test failed").into()),
			});
		core.run(f).unwrap();
	}

	#[test]
	fn test_server_http_codec() {
		let mut core = Core::new().unwrap();
		let request = "\
		               GET / HTTP/1.0\r\n\
		               Host: www.rust-lang.org\r\n\
		               \r\n\
		               "
		.as_bytes();
		let input = Cursor::new(request);
		let output = Cursor::new(Vec::new());

		let f = ReadWritePair(input, output)
			.framed(HttpServerCodec)
			.into_future()
			.map_err(|(e, _)| e)
			.and_then(|(m, s)| match m {
				Some(ref m) if m.subject.0 == Method::Get => Ok(s),
				_ => Err(io::Error::new(io::ErrorKind::Other, "test failed").into()),
			})
			.and_then(|s| {
				s.send(Incoming {
					version: HttpVersion::Http11,
					subject: StatusCode::NotFound,
					headers: Headers::new(),
				})
				.map_err(|e| e.into())
			});
		core.run(f).unwrap();
	}
}