actix/actix-extras
1
0
Fork 0
mirror of https://github.com/actix/actix-extras.git synced 2025-07-01 12:15:08 +02:00
Code Issues Releases Wiki Activity

add h2 server support

Browse Source
This commit is contained in:
Nikolay Kim
2019-02-06 11:44:15 -08:00
parent ef5b54a481
commit 55a29d3778
22 changed files with 774 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
src/body.rs
View File

@ -20,6 +20,18 @@ pub enum BodyLength {
Stream,
}
impl BodyLength {
pub fn is_eof(&self) -> bool {
match self {
BodyLength::None
| BodyLength::Empty
| BodyLength::Sized(0)
| BodyLength::Sized64(0) => true,
_ => false,
}
}
}
/// Type that provides this trait can be streamed to a peer.
pub trait MessageBody {
fn length(&self) -> BodyLength;
@ -42,6 +54,15 @@ pub enum ResponseBody<B> {
Other(Body),
}
impl ResponseBody<Body> {
pub fn into_body<B>(self) -> ResponseBody<B> {
match self {
ResponseBody::Body(b) => ResponseBody::Other(b),
ResponseBody::Other(b) => ResponseBody::Other(b),
}
}
}
impl<B: MessageBody> ResponseBody<B> {
pub fn as_ref(&self) -> Option<&B> {
if let ResponseBody::Body(ref b) = self {

11
src/client/connector.rs
View File

@ -12,11 +12,7 @@ use super::error::ConnectorError;
use super::pool::{ConnectionPool, Protocol};
#[cfg(feature = "ssl")]
use actix_connector::ssl::OpensslConnector;
#[cfg(feature = "ssl")]
use openssl::ssl::{SslConnector, SslMethod};
#[cfg(feature = "ssl")]
const H2: &[u8] = b"h2";
use openssl::ssl::SslConnector;
#[cfg(not(feature = "ssl"))]
type SslConnector = ();
@ -40,6 +36,8 @@ impl Default for Connector {
#[cfg(feature = "ssl")]
{
use log::error;
use openssl::ssl::{SslConnector, SslMethod};
let mut ssl = SslConnector::builder(SslMethod::tls()).unwrap();
let _ = ssl
.set_alpn_protos(b"\x02h2\x08http/1.1")
@ -167,6 +165,9 @@ impl Connector {
}
#[cfg(feature = "ssl")]
{
const H2: &[u8] = b"h2";
use actix_connector::ssl::OpensslConnector;
let ssl_service = Apply::new(
TimeoutService::new(self.timeout),
self.resolver

57
src/client/h2proto.rs
View File

@ -4,16 +4,19 @@ use std::time;
use actix_codec::{AsyncRead, AsyncWrite};
use bytes::Bytes;
use futures::future::{err, Either};
use futures::{Async, Future, Poll, Stream};
use futures::{Async, Future, Poll};
use h2::{client::SendRequest, SendStream};
use http::{request::Request, Version};
use http::header::{HeaderValue, CONNECTION, CONTENT_LENGTH, DATE, TRANSFER_ENCODING};
use http::{request::Request, HttpTryFrom, Version};
use crate::body::{BodyLength, MessageBody};
use crate::h2::Payload;
use crate::message::{RequestHead, ResponseHead};
use super::connection::{ConnectionType, IoConnection};
use super::error::SendRequestError;
use super::pool::Acquired;
use super::response::ClientResponse;
use crate::body::{BodyLength, MessageBody};
use crate::message::{RequestHead, ResponseHead};
pub(crate) fn send_request<T, B>(
io: SendRequest<Bytes>,
@ -27,7 +30,8 @@ where
B: MessageBody,
{
trace!("Sending client request: {:?} {:?}", head, body.length());
let eof = match body.length() {
let length = body.length();
let eof = match length {
BodyLength::None | BodyLength::Empty | BodyLength::Sized(0) => true,
_ => false,
};
@ -38,11 +42,44 @@ where
let mut req = Request::new(());
*req.uri_mut() = head.uri;
*req.method_mut() = head.method;
*req.headers_mut() = head.headers;
*req.version_mut() = Version::HTTP_2;
let mut skip_len = true;
let mut has_date = false;
// Content length
let _ = match length {
BodyLength::Chunked | BodyLength::None => None,
BodyLength::Stream => {
skip_len = false;
None
}
BodyLength::Empty => req
.headers_mut()
.insert(CONTENT_LENGTH, HeaderValue::from_static("0")),
BodyLength::Sized(len) => req.headers_mut().insert(
CONTENT_LENGTH,
HeaderValue::try_from(format!("{}", len)).unwrap(),
),
BodyLength::Sized64(len) => req.headers_mut().insert(
CONTENT_LENGTH,
HeaderValue::try_from(format!("{}", len)).unwrap(),
),
};
// copy headers
for (key, value) in head.headers.iter() {
match *key {
CONNECTION | TRANSFER_ENCODING => continue, // http2 specific
CONTENT_LENGTH if skip_len => continue,
DATE => has_date = true,
_ => (),
}
req.headers_mut().append(key, value.clone());
}
match io.send_request(req, eof) {
Ok((resp, send)) => {
Ok((res, send)) => {
release(io, pool, created, false);
if !eof {
@ -52,10 +89,10 @@ where
send,
buf: None,
}
.and_then(move |_| resp.map_err(SendRequestError::from)),
.and_then(move |_| res.map_err(SendRequestError::from)),
))
} else {
Either::B(resp.map_err(SendRequestError::from))
Either::B(res.map_err(SendRequestError::from))
}
}
Err(e) => {
@ -74,7 +111,7 @@ where
Ok(ClientResponse {
head,
payload: RefCell::new(Some(Box::new(body.from_err()))),
payload: RefCell::new(Some(Box::new(Payload::new(body)))),
})
})
.from_err()

2
src/client/response.rs
View File

@ -27,7 +27,7 @@ impl HttpMessage for ClientResponse {
}
#[inline]
fn payload(&self) -> Self::Stream {
fn payload(self) -> Self::Stream {
if let Some(payload) = self.payload.borrow_mut().take() {
payload
} else {

4
src/config.rs
View File

@ -171,6 +171,10 @@ impl ServiceConfig {
buf[35..].copy_from_slice(b"\r\n\r\n");
dst.extend_from_slice(&buf);
}
pub(crate) fn set_date_header(&self, dst: &mut BytesMut) {
dst.extend_from_slice(&self.0.timer.date().bytes);
}
}
/// A service config builder

10
src/error.rs
View File

@ -389,6 +389,10 @@ pub enum DispatchError<E: fmt::Debug> {
#[display(fmt = "Parse error: {}", _0)]
Parse(ParseError),
/// Http/2 error
#[display(fmt = "{}", _0)]
H2(h2::Error),
/// The first request did not complete within the specified timeout.
#[display(fmt = "The first request did not complete within the specified timeout")]
SlowRequestTimeout,
@ -426,6 +430,12 @@ impl<E: fmt::Debug> From<io::Error> for DispatchError<E> {
}
}
impl<E: fmt::Debug> From<h2::Error> for DispatchError<E> {
fn from(err: h2::Error) -> Self {
DispatchError::H2(err)
}
}
/// A set of error that can occure during parsing content type
#[derive(PartialEq, Debug, Display)]
pub enum ContentTypeError {

4
src/h1/dispatcher.rs
View File

@ -309,11 +309,11 @@ where
self.flags.insert(Flags::STARTED);
match msg {
Message::Item(req) => {
Message::Item(mut req) => {
match self.framed.get_codec().message_type() {
MessageType::Payload => {
let (ps, pl) = Payload::create(false);
*req.inner.payload.borrow_mut() = Some(pl);
req = req.set_payload(pl);
self.payload = Some(ps);
}
MessageType::Stream => {

325
src/h2/dispatcher.rs Normal file
View File

@ -0,0 +1,325 @@
use std::collections::VecDeque;
use std::marker::PhantomData;
use std::time::Instant;
use std::{fmt, mem};
use actix_codec::{AsyncRead, AsyncWrite};
use actix_service::Service;
use bitflags::bitflags;
use bytes::{Bytes, BytesMut};
use futures::{try_ready, Async, Future, Poll, Sink, Stream};
use h2::server::{Connection, SendResponse};
use h2::{RecvStream, SendStream};
use http::header::{
HeaderValue, ACCEPT_ENCODING, CONNECTION, CONTENT_LENGTH, DATE, TRANSFER_ENCODING,
};
use http::HttpTryFrom;
use log::{debug, error, trace};
use tokio_timer::Delay;
use crate::body::{Body, BodyLength, MessageBody, ResponseBody};
use crate::config::ServiceConfig;
use crate::error::{DispatchError, Error, ParseError, PayloadError, ResponseError};
use crate::message::ResponseHead;
use crate::request::Request;
use crate::response::Response;
use super::{H2ServiceResult, Payload};
const CHUNK_SIZE: usize = 16_384;
bitflags! {
struct Flags: u8 {
const DISCONNECTED = 0b0000_0001;
const SHUTDOWN = 0b0000_0010;
}
}
/// Dispatcher for HTTP/2 protocol
pub struct Dispatcher<T: AsyncRead + AsyncWrite, S: Service, B: MessageBody> {
flags: Flags,
service: S,
connection: Connection<T, Bytes>,
config: ServiceConfig,
ka_expire: Instant,
ka_timer: Option<Delay>,
_t: PhantomData<B>,
}
impl<T, S, B> Dispatcher<T, S, B>
where
T: AsyncRead + AsyncWrite,
S: Service<Request = Request<Payload>, Response = Response<B>> + 'static,
S::Error: Into<Error> + fmt::Debug,
B: MessageBody + 'static,
{
pub fn new(
service: S,
connection: Connection<T, Bytes>,
config: ServiceConfig,
timeout: Option<Delay>,
) -> Self {
let keepalive = config.keep_alive_enabled();
// let flags = if keepalive {
// Flags::KEEPALIVE | Flags::KEEPALIVE_ENABLED
// } else {
// Flags::empty()
// };
// keep-alive timer
let (ka_expire, ka_timer) = if let Some(delay) = timeout {
(delay.deadline(), Some(delay))
} else if let Some(delay) = config.keep_alive_timer() {
(delay.deadline(), Some(delay))
} else {
(config.now(), None)
};
Dispatcher {
service,
config,
ka_expire,
ka_timer,
connection,
flags: Flags::empty(),
_t: PhantomData,
}
}
}
impl<T, S, B> Future for Dispatcher<T, S, B>
where
T: AsyncRead + AsyncWrite,
S: Service<Request = Request<Payload>, Response = Response<B>> + 'static,
S::Error: Into<Error> + fmt::Debug,
B: MessageBody + 'static,
{
type Item = ();
type Error = DispatchError<()>;
#[inline]
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
loop {
match self.connection.poll()? {
Async::Ready(None) => {
self.flags.insert(Flags::DISCONNECTED);
}
Async::Ready(Some((req, res))) => {
// update keep-alive expire
if self.ka_timer.is_some() {
if let Some(expire) = self.config.keep_alive_expire() {
self.ka_expire = expire;
}
}
let (parts, body) = req.into_parts();
let mut req = Request::with_payload(Payload::new(body));
let head = &mut req.inner_mut().head;
head.uri = parts.uri;
head.method = parts.method;
head.version = parts.version;
head.headers = parts.headers;
tokio_current_thread::spawn(ServiceResponse::<S, B> {
state: ServiceResponseState::ServiceCall(
self.service.call(req),
Some(res),
),
config: self.config.clone(),
buffer: None,
})
}
Async::NotReady => return Ok(Async::NotReady),
}
}
}
}
struct ServiceResponse<S: Service, B> {
state: ServiceResponseState<S, B>,
config: ServiceConfig,
buffer: Option<Bytes>,
}
enum ServiceResponseState<S: Service, B> {
ServiceCall(S::Future, Option<SendResponse<Bytes>>),
SendPayload(SendStream<Bytes>, ResponseBody<B>),
}
impl<S, B> ServiceResponse<S, B>
where
S: Service<Request = Request<Payload>, Response = Response<B>> + 'static,
S::Error: Into<Error> + fmt::Debug,
B: MessageBody + 'static,
{
fn prepare_response(
&self,
head: &ResponseHead,
length: &mut BodyLength,
) -> http::Response<()> {
let mut has_date = false;
let mut skip_len = length != &BodyLength::Stream;
let mut res = http::Response::new(());
*res.status_mut() = head.status;
*res.version_mut() = http::Version::HTTP_2;
// Content length
match head.status {
http::StatusCode::NO_CONTENT
| http::StatusCode::CONTINUE
| http::StatusCode::PROCESSING => *length = BodyLength::None,
http::StatusCode::SWITCHING_PROTOCOLS => {
skip_len = true;
*length = BodyLength::Stream;
}
_ => (),
}
let _ = match length {
BodyLength::Chunked | BodyLength::None | BodyLength::Stream => None,
BodyLength::Empty => res
.headers_mut()
.insert(CONTENT_LENGTH, HeaderValue::from_static("0")),
BodyLength::Sized(len) => res.headers_mut().insert(
CONTENT_LENGTH,
HeaderValue::try_from(format!("{}", len)).unwrap(),
),
BodyLength::Sized64(len) => res.headers_mut().insert(
CONTENT_LENGTH,
HeaderValue::try_from(format!("{}", len)).unwrap(),
),
};
// copy headers
for (key, value) in head.headers.iter() {
match *key {
CONNECTION | TRANSFER_ENCODING => continue, // http2 specific
CONTENT_LENGTH if skip_len => continue,
DATE => has_date = true,
_ => (),
}
res.headers_mut().append(key, value.clone());
}
// set date header
if !has_date {
let mut bytes = BytesMut::with_capacity(29);
self.config.set_date_header(&mut bytes);
res.headers_mut()
.insert(DATE, HeaderValue::try_from(bytes.freeze()).unwrap());
}
res
}
}
impl<S, B> Future for ServiceResponse<S, B>
where
S: Service<Request = Request<Payload>, Response = Response<B>> + 'static,
S::Error: Into<Error> + fmt::Debug,
B: MessageBody + 'static,
{
type Item = ();
type Error = ();
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
match self.state {
ServiceResponseState::ServiceCall(ref mut call, ref mut send) => {
match call.poll() {
Ok(Async::Ready(res)) => {
let (res, body) = res.replace_body(());
let mut send = send.take().unwrap();
let mut length = body.length();
let h2_res = self.prepare_response(res.head(), &mut length);
let stream = send
.send_response(h2_res, length.is_eof())
.map_err(|e| {
trace!("Error sending h2 response: {:?}", e);
})?;
if length.is_eof() {
Ok(Async::Ready(()))
} else {
self.state = ServiceResponseState::SendPayload(stream, body);
self.poll()
}
}
Ok(Async::NotReady) => Ok(Async::NotReady),
Err(e) => {
let res: Response = e.into().into();
let (res, body) = res.replace_body(());
let mut send = send.take().unwrap();
let mut length = body.length();
let h2_res = self.prepare_response(res.head(), &mut length);
let stream = send
.send_response(h2_res, length.is_eof())
.map_err(|e| {
trace!("Error sending h2 response: {:?}", e);
})?;
if length.is_eof() {
Ok(Async::Ready(()))
} else {
self.state = ServiceResponseState::SendPayload(
stream,
body.into_body(),
);
self.poll()
}
}
}
}
ServiceResponseState::SendPayload(ref mut stream, ref mut body) => loop {
loop {
if let Some(ref mut buffer) = self.buffer {
match stream.poll_capacity().map_err(|e| warn!("{:?}", e))? {
Async::NotReady => return Ok(Async::NotReady),
Async::Ready(None) => return Ok(Async::Ready(())),
Async::Ready(Some(cap)) => {
let len = buffer.len();
let bytes = buffer.split_to(std::cmp::min(cap, len));
if let Err(e) = stream.send_data(bytes, false) {
warn!("{:?}", e);
return Err(());
} else if !buffer.is_empty() {
let cap = std::cmp::min(buffer.len(), CHUNK_SIZE);
stream.reserve_capacity(cap);
} else {
self.buffer.take();
}
}
}
} else {
match body.poll_next() {
Ok(Async::NotReady) => return Ok(Async::NotReady),
Ok(Async::Ready(None)) => {
if let Err(e) = stream.send_data(Bytes::new(), true) {
warn!("{:?}", e);
return Err(());
} else {
return Ok(Async::Ready(()));
}
}
Ok(Async::Ready(Some(chunk))) => {
stream.reserve_capacity(std::cmp::min(
chunk.len(),
CHUNK_SIZE,
));
self.buffer = Some(chunk);
}
Err(e) => {
error!("Response payload stream error: {:?}", e);
return Err(());
}
}
}
}
},
}
}
}

42
src/h2/mod.rs
View File

@ -1,7 +1,17 @@
#![allow(dead_code, unused_imports)]
use std::fmt;
use bytes::Bytes;
use futures::{Async, Poll, Stream};
use h2::RecvStream;
mod dispatcher;
mod service;
pub use self::service::H2Service;
use crate::error::PayloadError;
/// H1 service response type
pub enum H2ServiceResult<T> {
Disconnected,
@ -18,3 +28,35 @@ impl<T: fmt::Debug> fmt::Debug for H2ServiceResult<T> {
}
}
}
/// H2 receive stream
pub struct Payload {
pl: RecvStream,
}
impl Payload {
pub(crate) fn new(pl: RecvStream) -> Self {
Self { pl }
}
}
impl Stream for Payload {
type Item = Bytes;
type Error = PayloadError;
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
match self.pl.poll() {
Ok(Async::Ready(Some(chunk))) => {
let len = chunk.len();
if let Err(err) = self.pl.release_capacity().release_capacity(len) {
Err(err.into())
} else {
Ok(Async::Ready(Some(chunk)))
}
}
Ok(Async::Ready(None)) => Ok(Async::Ready(None)),
Ok(Async::NotReady) => Ok(Async::NotReady),
Err(err) => Err(err.into()),
}
}
}

136
src/h2/service.rs
View File

@ -1,6 +1,6 @@
use std::fmt::Debug;
use std::marker::PhantomData;
use std::net;
use std::{io, net};
use actix_codec::{AsyncRead, AsyncWrite, Framed};
use actix_service::{IntoNewService, NewService, Service};
@ -8,16 +8,17 @@ use bytes::Bytes;
use futures::future::{ok, FutureResult};
use futures::{try_ready, Async, Future, Poll, Stream};
use h2::server::{self, Connection, Handshake};
use h2::RecvStream;
use log::error;
use crate::body::MessageBody;
use crate::config::{KeepAlive, ServiceConfig};
use crate::error::{DispatchError, ParseError};
use crate::error::{DispatchError, Error, ParseError, ResponseError};
use crate::request::Request;
use crate::response::Response;
// use super::dispatcher::Dispatcher;
use super::H2ServiceResult;
use super::dispatcher::Dispatcher;
use super::{H2ServiceResult, Payload};
/// `NewService` implementation for HTTP2 transport
pub struct H2Service<T, S, B> {
@ -28,10 +29,10 @@ pub struct H2Service<T, S, B> {
impl<T, S, B> H2Service<T, S, B>
where
S: NewService<Request = Request, Response = Response<B>> + Clone,
S::Service: Clone,
S::Error: Debug,
B: MessageBody,
S: NewService<Request = Request<Payload>, Response = Response<B>> + Clone,
S::Service: Clone + 'static,
S::Error: Into<Error> + Debug + 'static,
B: MessageBody + 'static,
{
/// Create new `HttpService` instance.
pub fn new<F: IntoNewService<S>>(service: F) -> Self {
@ -53,14 +54,14 @@ where
impl<T, S, B> NewService for H2Service<T, S, B>
where
T: AsyncRead + AsyncWrite,
S: NewService<Request = Request, Response = Response<B>> + Clone,
S::Service: Clone,
S::Error: Debug,
B: MessageBody,
S: NewService<Request = Request<Payload>, Response = Response<B>> + Clone,
S::Service: Clone + 'static,
S::Error: Into<Error> + Debug,
B: MessageBody + 'static,
{
type Request = T;
type Response = H2ServiceResult<T>;
type Error = (); //DispatchError<S::Error>;
type Response = ();
type Error = DispatchError<()>;
type InitError = S::InitError;
type Service = H2ServiceHandler<T, S::Service, B>;
type Future = H2ServiceResponse<T, S, B>;
@ -90,9 +91,9 @@ pub struct H2ServiceBuilder<T, S> {
impl<T, S> H2ServiceBuilder<T, S>
where
S: NewService<Request = Request>,
S::Service: Clone,
S::Error: Debug,
S: NewService<Request = Request<Payload>>,
S::Service: Clone + 'static,
S::Error: Into<Error> + Debug + 'static,
{
/// Create instance of `H2ServiceBuilder`
pub fn new() -> H2ServiceBuilder<T, S> {
@ -185,6 +186,25 @@ where
self
}
// #[cfg(feature = "ssl")]
// /// Configure alpn protocols for SslAcceptorBuilder.
// pub fn configure_openssl(
// builder: &mut openssl::ssl::SslAcceptorBuilder,
// ) -> io::Result<()> {
// let protos: &[u8] = b"\x02h2";
// builder.set_alpn_select_callback(|_, protos| {
// const H2: &[u8] = b"\x02h2";
// if protos.windows(3).any(|window| window == H2) {
// Ok(b"h2")
// } else {
// Err(openssl::ssl::AlpnError::NOACK)
// }
// });
// builder.set_alpn_protos(&protos)?;
// Ok(())
// }
/// Finish service configuration and create `H1Service` instance.
pub fn finish<F, B>(self, service: F) -> H2Service<T, S, B>
where
@ -214,10 +234,10 @@ pub struct H2ServiceResponse<T, S: NewService, B> {
impl<T, S, B> Future for H2ServiceResponse<T, S, B>
where
T: AsyncRead + AsyncWrite,
S: NewService<Request = Request, Response = Response<B>>,
S::Service: Clone,
S::Error: Debug,
B: MessageBody,
S: NewService<Request = Request<Payload>, Response = Response<B>>,
S::Service: Clone + 'static,
S::Error: Into<Error> + Debug,
B: MessageBody + 'static,
{
type Item = H2ServiceHandler<T, S::Service, B>;
type Error = S::InitError;
@ -240,9 +260,9 @@ pub struct H2ServiceHandler<T, S, B> {
impl<T, S, B> H2ServiceHandler<T, S, B>
where
S: Service<Request = Request, Response = Response<B>> + Clone,
S::Error: Debug,
B: MessageBody,
S: Service<Request = Request<Payload>, Response = Response<B>> + Clone + 'static,
S::Error: Into<Error> + Debug,
B: MessageBody + 'static,
{
fn new(cfg: ServiceConfig, srv: S) -> H2ServiceHandler<T, S, B> {
H2ServiceHandler {
@ -256,55 +276,79 @@ where
impl<T, S, B> Service for H2ServiceHandler<T, S, B>
where
T: AsyncRead + AsyncWrite,
S: Service<Request = Request, Response = Response<B>> + Clone,
S::Error: Debug,
B: MessageBody,
S: Service<Request = Request<Payload>, Response = Response<B>> + Clone + 'static,
S::Error: Into<Error> + Debug,
B: MessageBody + 'static,
{
type Request = T;
type Response = H2ServiceResult<T>;
type Error = (); // DispatchError<S::Error>;
type Response = ();
type Error = DispatchError<()>;
type Future = H2ServiceHandlerResponse<T, S, B>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
self.srv.poll_ready().map_err(|_| ())
self.srv.poll_ready().map_err(|e| {
error!("Service readiness error: {:?}", e);
DispatchError::Service(())
})
}
fn call(&mut self, req: T) -> Self::Future {
H2ServiceHandlerResponse {
state: State::Handshake(server::handshake(req)),
_t: PhantomData,
state: State::Handshake(
Some(self.srv.clone()),
Some(self.cfg.clone()),
server::handshake(req),
),
}
}
}
enum State<T: AsyncRead + AsyncWrite> {
Handshake(Handshake<T, Bytes>),
Connection(Connection<T, Bytes>),
Empty,
enum State<T: AsyncRead + AsyncWrite, S: Service, B: MessageBody> {
Incoming(Dispatcher<T, S, B>),
Handshake(Option<S>, Option<ServiceConfig>, Handshake<T, Bytes>),
}
pub struct H2ServiceHandlerResponse<T, S, B>
where
T: AsyncRead + AsyncWrite,
S: Service<Request = Request, Response = Response<B>> + Clone,
S::Error: Debug,
B: MessageBody,
S: Service<Request = Request<Payload>, Response = Response<B>> + Clone + 'static,
S::Error: Into<Error> + Debug,
B: MessageBody + 'static,
{
state: State<T>,
_t: PhantomData<S>,
state: State<T, S, B>,
}
impl<T, S, B> Future for H2ServiceHandlerResponse<T, S, B>
where
T: AsyncRead + AsyncWrite,
S: Service<Request = Request, Response = Response<B>> + Clone,
S::Error: Debug,
S: Service<Request = Request<Payload>, Response = Response<B>> + Clone,
S::Error: Into<Error> + Debug,
B: MessageBody,
{
type Item = H2ServiceResult<T>;
type Error = ();
type Item = ();
type Error = DispatchError<()>;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
unimplemented!()
match self.state {
State::Incoming(ref mut disp) => disp.poll(),
State::Handshake(ref mut srv, ref mut config, ref mut handshake) => {
match handshake.poll() {
Ok(Async::Ready(conn)) => {
self.state = State::Incoming(Dispatcher::new(
srv.take().unwrap(),
conn,
config.take().unwrap(),
None,
));
self.poll()
}
Ok(Async::NotReady) => Ok(Async::NotReady),
Err(err) => {
trace!("H2 handshake error: {}", err);
return Err(err.into());
}
}
}
}
}
}

20
src/httpmessage.rs
View File

@ -25,7 +25,7 @@ pub trait HttpMessage: Sized {
fn headers(&self) -> &HeaderMap;
/// Message payload stream
fn payload(&self) -> Self::Stream;
fn payload(self) -> Self::Stream;
#[doc(hidden)]
/// Get a header
@ -128,7 +128,7 @@ pub trait HttpMessage: Sized {
/// }
/// # fn main() {}
/// ```
fn body(&self) -> MessageBody<Self> {
fn body(self) -> MessageBody<Self> {
MessageBody::new(self)
}
@ -162,7 +162,7 @@ pub trait HttpMessage: Sized {
/// }
/// # fn main() {}
/// ```
fn urlencoded<T: DeserializeOwned>(&self) -> UrlEncoded<Self, T> {
fn urlencoded<T: DeserializeOwned>(self) -> UrlEncoded<Self, T> {
UrlEncoded::new(self)
}
@ -198,12 +198,12 @@ pub trait HttpMessage: Sized {
/// }
/// # fn main() {}
/// ```
fn json<T: DeserializeOwned>(&self) -> JsonBody<Self, T> {
fn json<T: DeserializeOwned>(self) -> JsonBody<Self, T> {
JsonBody::new(self)
}
/// Return stream of lines.
fn readlines(&self) -> Readlines<Self> {
fn readlines(self) -> Readlines<Self> {
Readlines::new(self)
}
}
@ -220,7 +220,7 @@ pub struct Readlines<T: HttpMessage> {
impl<T: HttpMessage> Readlines<T> {
/// Create a new stream to read request line by line.
fn new(req: &T) -> Self {
fn new(req: T) -> Self {
let encoding = match req.encoding() {
Ok(enc) => enc,
Err(err) => return Self::err(req, err.into()),
@ -242,7 +242,7 @@ impl<T: HttpMessage> Readlines<T> {
self
}
fn err(req: &T, err: ReadlinesError) -> Self {
fn err(req: T, err: ReadlinesError) -> Self {
Readlines {
stream: req.payload(),
buff: BytesMut::new(),
@ -362,7 +362,7 @@ pub struct MessageBody<T: HttpMessage> {
impl<T: HttpMessage> MessageBody<T> {
/// Create `MessageBody` for request.
pub fn new(req: &T) -> MessageBody<T> {
pub fn new(req: T) -> MessageBody<T> {
let mut len = None;
if let Some(l) = req.headers().get(header::CONTENT_LENGTH) {
if let Ok(s) = l.to_str() {
@ -457,7 +457,7 @@ pub struct UrlEncoded<T: HttpMessage, U> {
impl<T: HttpMessage, U> UrlEncoded<T, U> {
/// Create a new future to URL encode a request
pub fn new(req: &T) -> UrlEncoded<T, U> {
pub fn new(req: T) -> UrlEncoded<T, U> {
// check content type
if req.content_type().to_lowercase() != "application/x-www-form-urlencoded" {
return Self::err(UrlencodedError::ContentType);
@ -800,7 +800,7 @@ mod tests {
Contrary to popular belief, Lorem Ipsum is not simply random text.",
))
.finish();
let mut r = Readlines::new(&req);
let mut r = Readlines::new(req);
match r.poll().ok().unwrap() {
Async::Ready(Some(s)) => assert_eq!(
s,

2
src/json.rs
View File

@ -50,7 +50,7 @@ pub struct JsonBody<T: HttpMessage, U: DeserializeOwned> {
impl<T: HttpMessage, U: DeserializeOwned> JsonBody<T, U> {
/// Create `JsonBody` for request.
pub fn new(req: &T) -> Self {
pub fn new(req: T) -> Self {
// check content-type
let json = if let Ok(Some(mime)) = req.mime_type() {
mime.subtype() == mime::JSON || mime.suffix() == Some(mime::JSON)

4
src/message.rs
View File

@ -5,7 +5,6 @@ use std::rc::Rc;
use http::{HeaderMap, Method, StatusCode, Uri, Version};
use crate::extensions::Extensions;
use crate::payload::Payload;
/// Represents various types of connection
#[derive(Copy, Clone, PartialEq, Debug)]
@ -149,7 +148,6 @@ impl ResponseHead {
pub struct Message<T: Head> {
pub head: T,
pub extensions: RefCell<Extensions>,
pub payload: RefCell<Option<Payload>>,
pub(crate) pool: &'static MessagePool<T>,
}
@ -159,7 +157,6 @@ impl<T: Head> Message<T> {
pub fn reset(&mut self) {
self.head.clear();
self.extensions.borrow_mut().clear();
*self.payload.borrow_mut() = None;
}
}
@ -168,7 +165,6 @@ impl<T: Head> Default for Message<T> {
Message {
pool: T::pool(),
head: T::default(),
payload: RefCell::new(None),
extensions: RefCell::new(Extensions::new()),
}
}

73
src/request.rs
View File

@ -2,43 +2,76 @@ use std::cell::{Ref, RefMut};
use std::fmt;
use std::rc::Rc;
use bytes::Bytes;
use futures::Stream;
use http::{header, HeaderMap, Method, Uri, Version};
use crate::error::PayloadError;
use crate::extensions::Extensions;
use crate::httpmessage::HttpMessage;
use crate::message::{Message, MessagePool, RequestHead};
use crate::payload::Payload;
/// Request
pub struct Request {
pub struct Request<P = Payload> {
pub(crate) payload: Option<P>,
pub(crate) inner: Rc<Message<RequestHead>>,
}
impl HttpMessage for Request {
type Stream = Payload;
impl<P> HttpMessage for Request<P>
where
P: Stream<Item = Bytes, Error = PayloadError>,
{
type Stream = P;
fn headers(&self) -> &HeaderMap {
&self.inner.head.headers
}
#[inline]
fn payload(&self) -> Payload {
if let Some(payload) = self.inner.payload.borrow_mut().take() {
payload
} else {
Payload::empty()
fn payload(mut self) -> P {
self.payload.take().unwrap()
}
}
impl Request<Payload> {
/// Create new Request instance
pub fn new() -> Request<Payload> {
Request {
payload: Some(Payload::empty()),
inner: MessagePool::get_message(),
}
}
}
impl Request {
impl<Payload> Request<Payload> {
/// Create new Request instance
pub fn new() -> Request {
pub fn with_payload(payload: Payload) -> Request<Payload> {
Request {
payload: Some(payload),
inner: MessagePool::get_message(),
}
}
/// Create new Request instance
pub fn set_payload<P>(self, payload: P) -> Request<P> {
Request {
payload: Some(payload),
inner: self.inner.clone(),
}
}
/// Take request's payload
pub fn take_payload(mut self) -> (Payload, Request<()>) {
(
self.payload.take().unwrap(),
Request {
payload: Some(()),
inner: self.inner.clone(),
},
)
}
// /// Create new Request instance with pool
// pub(crate) fn with_pool(pool: &'static MessagePool) -> Request {
// Request {
@ -143,17 +176,17 @@ impl Request {
self.inner().head.method == Method::CONNECT
}
#[doc(hidden)]
/// Note: this method should be called only as part of clone operation
/// of wrapper type.
pub fn clone_request(&self) -> Self {
Request {
inner: self.inner.clone(),
}
}
// #[doc(hidden)]
// /// Note: this method should be called only as part of clone operation
// /// of wrapper type.
// pub fn clone_request(&self) -> Self {
// Request {
// inner: self.inner.clone(),
// }
// }
}
impl Drop for Request {
impl<Payload> Drop for Request<Payload> {
fn drop(&mut self) {
if Rc::strong_count(&self.inner) == 1 {
self.inner.pool.release(self.inner.clone());
@ -161,7 +194,7 @@ impl Drop for Request {
}
}
impl fmt::Debug for Request {
impl<Payload> fmt::Debug for Request<Payload> {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
writeln!(
f,

21
src/test.rs
View File

@ -148,15 +148,18 @@ impl TestRequest {
..
} = self;
let mut req = Request::new();
{
let inner = req.inner_mut();
inner.head.uri = uri;
inner.head.method = method;
inner.head.version = version;
inner.head.headers = headers;
*inner.payload.borrow_mut() = payload;
}
let mut req = if let Some(pl) = payload {
Request::with_payload(pl)
} else {
Request::with_payload(Payload::empty())
};
let inner = req.inner_mut();
inner.head.uri = uri;
inner.head.method = method;
inner.head.version = version;
inner.head.headers = headers;
// req.set_cookies(cookies);
req
}