feat(server): Make the server code an optional feature (#2334)
cc #2223 BREAKING CHANGE: The HTTP server code is now an optional feature. To enable the server, add `features = ["server"]` to the dependency in your `Cargo.toml`.
This commit is contained in:
@@ -75,6 +75,7 @@ default = [
|
||||
"runtime",
|
||||
"stream",
|
||||
"client",
|
||||
"server",
|
||||
"http1",
|
||||
"http2",
|
||||
]
|
||||
@@ -82,6 +83,7 @@ full = [
|
||||
"client",
|
||||
"http1",
|
||||
"http2",
|
||||
"server",
|
||||
"stream",
|
||||
"runtime",
|
||||
]
|
||||
@@ -100,7 +102,9 @@ tcp = [
|
||||
http1 = []
|
||||
http2 = ["h2"]
|
||||
|
||||
# Client/Server
|
||||
client = []
|
||||
server = []
|
||||
|
||||
# `impl Stream` for things
|
||||
stream = []
|
||||
|
||||
@@ -21,7 +21,7 @@ use crate::common::{task, watch, Pin, Poll};
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
#[cfg(feature = "client")]
|
||||
use crate::common::{Future, Never};
|
||||
#[cfg(feature = "http2")]
|
||||
#[cfg(all(feature = "http2", any(feature = "client", feature = "server")))]
|
||||
use crate::proto::h2::ping;
|
||||
use crate::upgrade::OnUpgrade;
|
||||
|
||||
@@ -46,7 +46,7 @@ enum Kind {
|
||||
want_tx: watch::Sender,
|
||||
rx: mpsc::Receiver<Result<Bytes, crate::Error>>,
|
||||
},
|
||||
#[cfg(feature = "http2")]
|
||||
#[cfg(all(feature = "http2", any(feature = "client", feature = "server")))]
|
||||
H2 {
|
||||
ping: ping::Recorder,
|
||||
content_length: DecodedLength,
|
||||
@@ -200,7 +200,7 @@ impl Body {
|
||||
Body { kind, extra: None }
|
||||
}
|
||||
|
||||
#[cfg(feature = "http2")]
|
||||
#[cfg(all(feature = "http2", any(feature = "client", feature = "server")))]
|
||||
pub(crate) fn h2(
|
||||
recv: h2::RecvStream,
|
||||
content_length: DecodedLength,
|
||||
@@ -301,7 +301,7 @@ impl Body {
|
||||
None => Poll::Ready(None),
|
||||
}
|
||||
}
|
||||
#[cfg(feature = "http2")]
|
||||
#[cfg(all(feature = "http2", any(feature = "client", feature = "server")))]
|
||||
Kind::H2 {
|
||||
ref ping,
|
||||
recv: ref mut h2,
|
||||
@@ -359,7 +359,7 @@ impl HttpBody for Body {
|
||||
#[cfg_attr(not(feature = "http2"), allow(unused))] cx: &mut task::Context<'_>,
|
||||
) -> Poll<Result<Option<HeaderMap>, Self::Error>> {
|
||||
match self.kind {
|
||||
#[cfg(feature = "http2")]
|
||||
#[cfg(all(feature = "http2", any(feature = "client", feature = "server")))]
|
||||
Kind::H2 {
|
||||
recv: ref mut h2,
|
||||
ref ping,
|
||||
@@ -379,7 +379,7 @@ impl HttpBody for Body {
|
||||
match self.kind {
|
||||
Kind::Once(ref val) => val.is_none(),
|
||||
Kind::Chan { content_length, .. } => content_length == DecodedLength::ZERO,
|
||||
#[cfg(feature = "http2")]
|
||||
#[cfg(all(feature = "http2", any(feature = "client", feature = "server")))]
|
||||
Kind::H2 { recv: ref h2, .. } => h2.is_end_stream(),
|
||||
#[cfg(feature = "stream")]
|
||||
Kind::Wrapped(..) => false,
|
||||
@@ -405,7 +405,7 @@ impl HttpBody for Body {
|
||||
#[cfg(feature = "stream")]
|
||||
Kind::Wrapped(..) => SizeHint::default(),
|
||||
Kind::Chan { content_length, .. } => opt_len!(content_length),
|
||||
#[cfg(feature = "http2")]
|
||||
#[cfg(all(feature = "http2", any(feature = "client", feature = "server")))]
|
||||
Kind::H2 { content_length, .. } => opt_len!(content_length),
|
||||
}
|
||||
}
|
||||
|
||||
18
src/cfg.rs
18
src/cfg.rs
@@ -11,16 +11,19 @@ macro_rules! cfg_feature {
|
||||
}
|
||||
}
|
||||
|
||||
macro_rules! cfg_any_http {
|
||||
macro_rules! cfg_proto {
|
||||
($($item:item)*) => {
|
||||
cfg_feature! {
|
||||
#![any(feature = "http1", feature = "http2")]
|
||||
#![all(
|
||||
any(feature = "http1", feature = "http2"),
|
||||
any(feature = "client", feature = "server"),
|
||||
)]
|
||||
$($item)*
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
cfg_any_http! {
|
||||
cfg_proto! {
|
||||
macro_rules! cfg_http1 {
|
||||
($($item:item)*) => {
|
||||
cfg_feature! {
|
||||
@@ -47,4 +50,13 @@ cfg_any_http! {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
macro_rules! cfg_server {
|
||||
($($item:item)*) => {
|
||||
cfg_feature! {
|
||||
#![feature = "server"]
|
||||
$($item)*
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,17 +3,23 @@ use std::future::Future;
|
||||
use std::pin::Pin;
|
||||
use std::sync::Arc;
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
use crate::body::{Body, HttpBody};
|
||||
#[cfg(feature = "http2")]
|
||||
#[cfg(feature = "server")]
|
||||
use crate::proto::h2::server::H2Stream;
|
||||
use crate::rt::Executor;
|
||||
#[cfg(feature = "server")]
|
||||
use crate::server::conn::spawn_all::{NewSvcTask, Watcher};
|
||||
#[cfg(feature = "server")]
|
||||
use crate::service::HttpService;
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
pub trait ConnStreamExec<F, B: HttpBody>: Clone {
|
||||
fn execute_h2stream(&mut self, fut: H2Stream<F, B>);
|
||||
}
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
pub trait NewSvcExec<I, N, S: HttpService<Body>, E, W: Watcher<I, S, E>>: Clone {
|
||||
fn execute_new_svc(&mut self, fut: NewSvcTask<I, N, S, E, W>);
|
||||
}
|
||||
@@ -60,6 +66,7 @@ impl fmt::Debug for Exec {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
impl<F, B> ConnStreamExec<F, B> for Exec
|
||||
where
|
||||
H2Stream<F, B>: Future<Output = ()> + Send + 'static,
|
||||
@@ -70,6 +77,7 @@ where
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
impl<I, N, S, E, W> NewSvcExec<I, N, S, E, W> for Exec
|
||||
where
|
||||
NewSvcTask<I, N, S, E, W>: Future<Output = ()> + Send + 'static,
|
||||
@@ -83,6 +91,7 @@ where
|
||||
|
||||
// ==== impl Executor =====
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
impl<E, F, B> ConnStreamExec<F, B> for E
|
||||
where
|
||||
E: Executor<H2Stream<F, B>> + Clone,
|
||||
@@ -94,6 +103,7 @@ where
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
impl<I, N, S, E, W> NewSvcExec<I, N, S, E, W> for E
|
||||
where
|
||||
E: Executor<NewSvcTask<I, N, S, E, W>> + Clone,
|
||||
@@ -119,7 +129,7 @@ pub struct H2Stream<F, B>(std::marker::PhantomData<(F, B)>);
|
||||
impl<F, B, E> Future for H2Stream<F, B>
|
||||
where
|
||||
F: Future<Output = Result<http::Response<B>, E>>,
|
||||
B: HttpBody,
|
||||
B: crate::body::HttpBody,
|
||||
B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
|
||||
E: Into<Box<dyn std::error::Error + Send + Sync>>,
|
||||
{
|
||||
|
||||
@@ -15,6 +15,7 @@ pub(crate) struct Rewind<T> {
|
||||
|
||||
impl<T> Rewind<T> {
|
||||
#[cfg(any(feature = "http2", test))]
|
||||
#[cfg(feature = "server")]
|
||||
pub(crate) fn new(io: T) -> Self {
|
||||
Rewind {
|
||||
pre: None,
|
||||
@@ -29,7 +30,7 @@ impl<T> Rewind<T> {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(any(all(feature = "http1", feature = "http2"), test))]
|
||||
#[cfg(any(all(feature = "http1", feature = "http2", feature = "server"), test))]
|
||||
pub(crate) fn rewind(&mut self, bs: Bytes) {
|
||||
debug_assert!(self.pre.is_none());
|
||||
self.pre = Some(bs);
|
||||
|
||||
@@ -9,8 +9,10 @@ macro_rules! ready {
|
||||
|
||||
pub(crate) mod buf;
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
#[cfg(feature = "server")]
|
||||
pub(crate) mod date;
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
#[cfg(feature = "server")]
|
||||
pub(crate) mod drain;
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
pub(crate) mod exec;
|
||||
@@ -24,8 +26,6 @@ pub(crate) mod sync_wrapper;
|
||||
pub(crate) mod task;
|
||||
pub(crate) mod watch;
|
||||
|
||||
//#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
//pub(crate) use self::exec::{BoxSendFuture, Exec};
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
#[cfg(feature = "client")]
|
||||
pub(crate) use self::lazy::{lazy, Started as Lazy};
|
||||
@@ -33,6 +33,7 @@ pub use self::never::Never;
|
||||
pub(crate) use self::task::Poll;
|
||||
|
||||
// group up types normally needed for `Future`
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
pub(crate) use std::marker::Unpin;
|
||||
cfg_proto! {
|
||||
pub(crate) use std::marker::Unpin;
|
||||
}
|
||||
pub(crate) use std::{future::Future, pin::Pin};
|
||||
|
||||
21
src/error.rs
21
src/error.rs
@@ -36,10 +36,15 @@ pub(crate) enum Kind {
|
||||
/// Error occurred while connecting.
|
||||
Connect,
|
||||
/// Error creating a TcpListener.
|
||||
#[cfg(all(any(feature = "http1", feature = "http2"), feature = "tcp"))]
|
||||
#[cfg(all(
|
||||
any(feature = "http1", feature = "http2"),
|
||||
feature = "tcp",
|
||||
feature = "server"
|
||||
))]
|
||||
Listen,
|
||||
/// Error accepting on an Incoming stream.
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
#[cfg(feature = "server")]
|
||||
Accept,
|
||||
/// Error while reading a body from connection.
|
||||
#[cfg(any(feature = "http1", feature = "http2", feature = "stream"))]
|
||||
@@ -77,6 +82,7 @@ pub(crate) enum User {
|
||||
Body,
|
||||
/// Error calling user's MakeService.
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
#[cfg(feature = "server")]
|
||||
MakeService,
|
||||
/// Error from future of user's Service.
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
@@ -85,6 +91,7 @@ pub(crate) enum User {
|
||||
///
|
||||
/// For example, sending both `content-length` and `transfer-encoding`.
|
||||
#[cfg(feature = "http1")]
|
||||
#[cfg(feature = "server")]
|
||||
UnexpectedHeader,
|
||||
/// User tried to create a Request with bad version.
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
@@ -96,6 +103,7 @@ pub(crate) enum User {
|
||||
UnsupportedRequestMethod,
|
||||
/// User tried to respond with a 1xx (not 101) response code.
|
||||
#[cfg(feature = "http1")]
|
||||
#[cfg(feature = "server")]
|
||||
UnsupportedStatusCode,
|
||||
/// User tried to send a Request with Client with non-absolute URI.
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
@@ -178,6 +186,7 @@ impl Error {
|
||||
}
|
||||
|
||||
#[cfg(feature = "http1")]
|
||||
#[cfg(feature = "server")]
|
||||
pub(crate) fn kind(&self) -> &Kind {
|
||||
&self.inner.kind
|
||||
}
|
||||
@@ -234,11 +243,13 @@ impl Error {
|
||||
}
|
||||
|
||||
#[cfg(all(any(feature = "http1", feature = "http2"), feature = "tcp"))]
|
||||
#[cfg(feature = "server")]
|
||||
pub(crate) fn new_listen<E: Into<Cause>>(cause: E) -> Error {
|
||||
Error::new(Kind::Listen).with(cause)
|
||||
}
|
||||
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
#[cfg(feature = "server")]
|
||||
pub(crate) fn new_accept<E: Into<Cause>>(cause: E) -> Error {
|
||||
Error::new(Kind::Accept).with(cause)
|
||||
}
|
||||
@@ -272,6 +283,7 @@ impl Error {
|
||||
}
|
||||
|
||||
#[cfg(feature = "http1")]
|
||||
#[cfg(feature = "server")]
|
||||
pub(crate) fn new_user_header() -> Error {
|
||||
Error::new_user(User::UnexpectedHeader)
|
||||
}
|
||||
@@ -289,6 +301,7 @@ impl Error {
|
||||
}
|
||||
|
||||
#[cfg(feature = "http1")]
|
||||
#[cfg(feature = "server")]
|
||||
pub(crate) fn new_user_unsupported_status_code() -> Error {
|
||||
Error::new_user(User::UnsupportedStatusCode)
|
||||
}
|
||||
@@ -309,6 +322,7 @@ impl Error {
|
||||
}
|
||||
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
#[cfg(feature = "server")]
|
||||
pub(crate) fn new_user_make_service<E: Into<Cause>>(cause: E) -> Error {
|
||||
Error::new_user(User::MakeService).with(cause)
|
||||
}
|
||||
@@ -354,8 +368,10 @@ impl Error {
|
||||
Kind::Connect => "error trying to connect",
|
||||
Kind::Canceled => "operation was canceled",
|
||||
#[cfg(all(any(feature = "http1", feature = "http2"), feature = "tcp"))]
|
||||
#[cfg(feature = "server")]
|
||||
Kind::Listen => "error creating server listener",
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
#[cfg(feature = "server")]
|
||||
Kind::Accept => "error accepting connection",
|
||||
#[cfg(any(feature = "http1", feature = "http2", feature = "stream"))]
|
||||
Kind::Body => "error reading a body from connection",
|
||||
@@ -372,10 +388,12 @@ impl Error {
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
Kind::User(User::Body) => "error from user's HttpBody stream",
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
#[cfg(feature = "server")]
|
||||
Kind::User(User::MakeService) => "error from user's MakeService",
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
Kind::User(User::Service) => "error from user's Service",
|
||||
#[cfg(feature = "http1")]
|
||||
#[cfg(feature = "server")]
|
||||
Kind::User(User::UnexpectedHeader) => "user sent unexpected header",
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
#[cfg(feature = "client")]
|
||||
@@ -384,6 +402,7 @@ impl Error {
|
||||
#[cfg(feature = "client")]
|
||||
Kind::User(User::UnsupportedRequestMethod) => "request has unsupported HTTP method",
|
||||
#[cfg(feature = "http1")]
|
||||
#[cfg(feature = "server")]
|
||||
Kind::User(User::UnsupportedStatusCode) => {
|
||||
"response has 1xx status code, not supported by server"
|
||||
}
|
||||
|
||||
@@ -30,6 +30,7 @@ fn connection_has(value: &HeaderValue, needle: &str) -> bool {
|
||||
}
|
||||
|
||||
#[cfg(feature = "http1")]
|
||||
#[cfg(feature = "server")]
|
||||
pub fn content_length_parse(value: &HeaderValue) -> Option<u64> {
|
||||
value.to_str().ok().and_then(|s| s.parse().ok())
|
||||
}
|
||||
|
||||
12
src/lib.rs
12
src/lib.rs
@@ -68,12 +68,9 @@ pub mod rt;
|
||||
pub mod service;
|
||||
pub mod upgrade;
|
||||
|
||||
cfg_any_http! {
|
||||
cfg_proto! {
|
||||
mod headers;
|
||||
mod proto;
|
||||
pub mod server;
|
||||
|
||||
pub use crate::server::Server;
|
||||
}
|
||||
|
||||
cfg_feature! {
|
||||
@@ -82,3 +79,10 @@ cfg_feature! {
|
||||
pub mod client;
|
||||
pub use crate::client::Client;
|
||||
}
|
||||
|
||||
cfg_feature! {
|
||||
#![all(feature = "server", any(feature = "http1", feature = "http2"))]
|
||||
|
||||
pub mod server;
|
||||
pub use crate::server::Server;
|
||||
}
|
||||
|
||||
@@ -57,6 +57,7 @@ where
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
pub fn set_flush_pipeline(&mut self, enabled: bool) {
|
||||
self.io.set_flush_pipeline(enabled);
|
||||
}
|
||||
@@ -83,6 +84,7 @@ where
|
||||
self.state.title_case_headers = true;
|
||||
}
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
pub(crate) fn set_allow_half_close(&mut self) {
|
||||
self.state.allow_half_close = true;
|
||||
}
|
||||
@@ -485,6 +487,7 @@ where
|
||||
Encode {
|
||||
head: &mut head,
|
||||
body,
|
||||
#[cfg(feature = "server")]
|
||||
keep_alive: self.state.wants_keep_alive(),
|
||||
req_method: &mut self.state.method,
|
||||
title_case_headers: self.state.title_case_headers,
|
||||
@@ -690,6 +693,7 @@ where
|
||||
self.state.close_write();
|
||||
}
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
pub fn disable_keep_alive(&mut self) {
|
||||
if self.state.is_idle() {
|
||||
trace!("disable_keep_alive; closing idle connection");
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
use std::error::Error as StdError;
|
||||
|
||||
use bytes::{Buf, Bytes};
|
||||
use http::{Request, StatusCode};
|
||||
use http::Request;
|
||||
use tokio::io::{AsyncRead, AsyncWrite};
|
||||
|
||||
use super::{Http1Transaction, Wants};
|
||||
@@ -10,7 +10,6 @@ use crate::common::{task, Future, Pin, Poll, Unpin};
|
||||
use crate::proto::{
|
||||
BodyLength, Conn, Dispatched, MessageHead, RequestHead,
|
||||
};
|
||||
use crate::service::HttpService;
|
||||
|
||||
pub(crate) struct Dispatcher<D, Bs: HttpBody, I, T> {
|
||||
conn: Conn<I, Bs::Data, T>,
|
||||
@@ -34,9 +33,13 @@ pub(crate) trait Dispatch {
|
||||
fn should_poll(&self) -> bool;
|
||||
}
|
||||
|
||||
pub struct Server<S: HttpService<B>, B> {
|
||||
in_flight: Pin<Box<Option<S::Future>>>,
|
||||
pub(crate) service: S,
|
||||
cfg_server! {
|
||||
use crate::service::HttpService;
|
||||
|
||||
pub struct Server<S: HttpService<B>, B> {
|
||||
in_flight: Pin<Box<Option<S::Future>>>,
|
||||
pub(crate) service: S,
|
||||
}
|
||||
}
|
||||
|
||||
cfg_client! {
|
||||
@@ -74,6 +77,7 @@ where
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
pub fn disable_keep_alive(&mut self) {
|
||||
self.conn.disable_keep_alive();
|
||||
if self.conn.is_write_closed() {
|
||||
@@ -441,84 +445,86 @@ impl<'a, T> Drop for OptGuard<'a, T> {
|
||||
|
||||
// ===== impl Server =====
|
||||
|
||||
impl<S, B> Server<S, B>
|
||||
where
|
||||
S: HttpService<B>,
|
||||
{
|
||||
pub fn new(service: S) -> Server<S, B> {
|
||||
Server {
|
||||
in_flight: Box::pin(None),
|
||||
service,
|
||||
cfg_server! {
|
||||
impl<S, B> Server<S, B>
|
||||
where
|
||||
S: HttpService<B>,
|
||||
{
|
||||
pub fn new(service: S) -> Server<S, B> {
|
||||
Server {
|
||||
in_flight: Box::pin(None),
|
||||
service,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn into_service(self) -> S {
|
||||
self.service
|
||||
}
|
||||
}
|
||||
|
||||
pub fn into_service(self) -> S {
|
||||
self.service
|
||||
}
|
||||
}
|
||||
// Service is never pinned
|
||||
impl<S: HttpService<B>, B> Unpin for Server<S, B> {}
|
||||
|
||||
// Service is never pinned
|
||||
impl<S: HttpService<B>, B> Unpin for Server<S, B> {}
|
||||
impl<S, Bs> Dispatch for Server<S, Body>
|
||||
where
|
||||
S: HttpService<Body, ResBody = Bs>,
|
||||
S::Error: Into<Box<dyn StdError + Send + Sync>>,
|
||||
Bs: HttpBody,
|
||||
{
|
||||
type PollItem = MessageHead<http::StatusCode>;
|
||||
type PollBody = Bs;
|
||||
type PollError = S::Error;
|
||||
type RecvItem = RequestHead;
|
||||
|
||||
impl<S, Bs> Dispatch for Server<S, Body>
|
||||
where
|
||||
S: HttpService<Body, ResBody = Bs>,
|
||||
S::Error: Into<Box<dyn StdError + Send + Sync>>,
|
||||
Bs: HttpBody,
|
||||
{
|
||||
type PollItem = MessageHead<StatusCode>;
|
||||
type PollBody = Bs;
|
||||
type PollError = S::Error;
|
||||
type RecvItem = RequestHead;
|
||||
|
||||
fn poll_msg(
|
||||
mut self: Pin<&mut Self>,
|
||||
cx: &mut task::Context<'_>,
|
||||
) -> Poll<Option<Result<(Self::PollItem, Self::PollBody), Self::PollError>>> {
|
||||
let mut this = self.as_mut();
|
||||
let ret = if let Some(ref mut fut) = this.in_flight.as_mut().as_pin_mut() {
|
||||
let resp = ready!(fut.as_mut().poll(cx)?);
|
||||
let (parts, body) = resp.into_parts();
|
||||
let head = MessageHead {
|
||||
version: parts.version,
|
||||
subject: parts.status,
|
||||
headers: parts.headers,
|
||||
fn poll_msg(
|
||||
mut self: Pin<&mut Self>,
|
||||
cx: &mut task::Context<'_>,
|
||||
) -> Poll<Option<Result<(Self::PollItem, Self::PollBody), Self::PollError>>> {
|
||||
let mut this = self.as_mut();
|
||||
let ret = if let Some(ref mut fut) = this.in_flight.as_mut().as_pin_mut() {
|
||||
let resp = ready!(fut.as_mut().poll(cx)?);
|
||||
let (parts, body) = resp.into_parts();
|
||||
let head = MessageHead {
|
||||
version: parts.version,
|
||||
subject: parts.status,
|
||||
headers: parts.headers,
|
||||
};
|
||||
Poll::Ready(Some(Ok((head, body))))
|
||||
} else {
|
||||
unreachable!("poll_msg shouldn't be called if no inflight");
|
||||
};
|
||||
Poll::Ready(Some(Ok((head, body))))
|
||||
} else {
|
||||
unreachable!("poll_msg shouldn't be called if no inflight");
|
||||
};
|
||||
|
||||
// Since in_flight finished, remove it
|
||||
this.in_flight.set(None);
|
||||
ret
|
||||
}
|
||||
|
||||
fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, Body)>) -> crate::Result<()> {
|
||||
let (msg, body) = msg?;
|
||||
let mut req = Request::new(body);
|
||||
*req.method_mut() = msg.subject.0;
|
||||
*req.uri_mut() = msg.subject.1;
|
||||
*req.headers_mut() = msg.headers;
|
||||
*req.version_mut() = msg.version;
|
||||
let fut = self.service.call(req);
|
||||
self.in_flight.set(Some(fut));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), ()>> {
|
||||
if self.in_flight.is_some() {
|
||||
Poll::Pending
|
||||
} else {
|
||||
self.service.poll_ready(cx).map_err(|_e| {
|
||||
// FIXME: return error value.
|
||||
trace!("service closed");
|
||||
})
|
||||
// Since in_flight finished, remove it
|
||||
this.in_flight.set(None);
|
||||
ret
|
||||
}
|
||||
}
|
||||
|
||||
fn should_poll(&self) -> bool {
|
||||
self.in_flight.is_some()
|
||||
fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, Body)>) -> crate::Result<()> {
|
||||
let (msg, body) = msg?;
|
||||
let mut req = Request::new(body);
|
||||
*req.method_mut() = msg.subject.0;
|
||||
*req.uri_mut() = msg.subject.1;
|
||||
*req.headers_mut() = msg.headers;
|
||||
*req.version_mut() = msg.version;
|
||||
let fut = self.service.call(req);
|
||||
self.in_flight.set(Some(fut));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), ()>> {
|
||||
if self.in_flight.is_some() {
|
||||
Poll::Pending
|
||||
} else {
|
||||
self.service.poll_ready(cx).map_err(|_e| {
|
||||
// FIXME: return error value.
|
||||
trace!("service closed");
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
fn should_poll(&self) -> bool {
|
||||
self.in_flight.is_some()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -35,6 +35,7 @@ enum Kind {
|
||||
///
|
||||
/// This is mostly only used with HTTP/1.0 with a length. This kind requires
|
||||
/// the connection to be closed when the body is finished.
|
||||
#[cfg(feature = "server")]
|
||||
CloseDelimited,
|
||||
}
|
||||
|
||||
@@ -61,6 +62,7 @@ impl Encoder {
|
||||
Encoder::new(Kind::Length(len))
|
||||
}
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
pub fn close_delimited() -> Encoder {
|
||||
Encoder::new(Kind::CloseDelimited)
|
||||
}
|
||||
@@ -72,6 +74,7 @@ impl Encoder {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
pub fn set_last(mut self, is_last: bool) -> Self {
|
||||
self.is_last = is_last;
|
||||
self
|
||||
@@ -83,6 +86,7 @@ impl Encoder {
|
||||
|
||||
pub fn is_close_delimited(&self) -> bool {
|
||||
match self.kind {
|
||||
#[cfg(feature = "server")]
|
||||
Kind::CloseDelimited => true,
|
||||
_ => false,
|
||||
}
|
||||
@@ -94,6 +98,7 @@ impl Encoder {
|
||||
Kind::Chunked => Ok(Some(EncodedBuf {
|
||||
kind: BufKind::ChunkedEnd(b"0\r\n\r\n"),
|
||||
})),
|
||||
#[cfg(feature = "server")]
|
||||
Kind::CloseDelimited => Ok(None),
|
||||
Kind::Length(_) => Err(NotEof),
|
||||
}
|
||||
@@ -125,6 +130,7 @@ impl Encoder {
|
||||
BufKind::Exact(msg)
|
||||
}
|
||||
}
|
||||
#[cfg(feature = "server")]
|
||||
Kind::CloseDelimited => {
|
||||
trace!("close delimited write {}B", len);
|
||||
BufKind::Exact(msg)
|
||||
@@ -168,6 +174,7 @@ impl Encoder {
|
||||
}
|
||||
}
|
||||
}
|
||||
#[cfg(feature = "server")]
|
||||
Kind::CloseDelimited => {
|
||||
trace!("close delimited write {}B", len);
|
||||
dst.buffer(msg);
|
||||
|
||||
@@ -66,6 +66,7 @@ where
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
pub fn set_flush_pipeline(&mut self, enabled: bool) {
|
||||
debug_assert!(!self.write_buf.has_remaining());
|
||||
self.flush_pipeline = enabled;
|
||||
|
||||
@@ -18,12 +18,15 @@ mod encode;
|
||||
mod io;
|
||||
mod role;
|
||||
|
||||
pub(crate) type ServerTransaction = role::Server;
|
||||
|
||||
cfg_client! {
|
||||
pub(crate) type ClientTransaction = role::Client;
|
||||
}
|
||||
|
||||
cfg_server! {
|
||||
pub(crate) type ServerTransaction = role::Server;
|
||||
}
|
||||
|
||||
pub(crate) trait Http1Transaction {
|
||||
type Incoming;
|
||||
type Outgoing: Default;
|
||||
@@ -73,6 +76,7 @@ pub(crate) struct ParseContext<'a> {
|
||||
pub(crate) struct Encode<'a, T> {
|
||||
head: &'a mut MessageHead<T>,
|
||||
body: Option<BodyLength>,
|
||||
#[cfg(feature = "server")]
|
||||
keep_alive: bool,
|
||||
req_method: &'a mut Option<Method>,
|
||||
title_case_headers: bool,
|
||||
|
||||
@@ -10,6 +10,7 @@ use http::header::{self, Entry, HeaderName, HeaderValue};
|
||||
use http::{HeaderMap, Method, StatusCode, Version};
|
||||
|
||||
use crate::body::DecodedLength;
|
||||
#[cfg(feature = "server")]
|
||||
use crate::common::date;
|
||||
use crate::error::Parse;
|
||||
use crate::headers;
|
||||
@@ -94,10 +95,13 @@ where
|
||||
|
||||
// There are 2 main roles, Client and Server.
|
||||
|
||||
#[cfg(feature = "client")]
|
||||
pub(crate) enum Client {}
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
pub(crate) enum Server {}
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
impl Http1Transaction for Server {
|
||||
type Incoming = RequestLine;
|
||||
type Outgoing = StatusCode;
|
||||
@@ -618,6 +622,7 @@ impl Http1Transaction for Server {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
impl Server {
|
||||
fn can_have_body(method: &Option<Method>, status: StatusCode) -> bool {
|
||||
Server::can_chunked(method, status)
|
||||
@@ -640,6 +645,7 @@ impl Server {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "client")]
|
||||
impl Http1Transaction for Client {
|
||||
type Incoming = StatusCode;
|
||||
type Outgoing = RequestLine;
|
||||
@@ -779,6 +785,7 @@ impl Http1Transaction for Client {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "client")]
|
||||
impl Client {
|
||||
/// Returns Some(length, wants_upgrade) if successful.
|
||||
///
|
||||
@@ -846,9 +853,6 @@ impl Client {
|
||||
Ok(Some((DecodedLength::CLOSE_DELIMITED, false)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Client {
|
||||
fn set_length(head: &mut RequestHead, body: Option<BodyLength>) -> Encoder {
|
||||
let body = if let Some(body) = body {
|
||||
body
|
||||
|
||||
@@ -14,14 +14,16 @@ use crate::common::{task, Future, Pin, Poll};
|
||||
use crate::headers::content_length_parse_all;
|
||||
|
||||
pub(crate) mod ping;
|
||||
pub(crate) mod server;
|
||||
|
||||
cfg_client! {
|
||||
pub(crate) mod client;
|
||||
pub(crate) use self::client::ClientTask;
|
||||
}
|
||||
|
||||
pub(crate) use self::server::Server;
|
||||
cfg_server! {
|
||||
pub(crate) mod server;
|
||||
pub(crate) use self::server::Server;
|
||||
}
|
||||
|
||||
/// Default initial stream window size defined in HTTP2 spec.
|
||||
pub(crate) const SPEC_WINDOW_SIZE: u32 = 65_535;
|
||||
|
||||
@@ -3,10 +3,12 @@
|
||||
cfg_http1! {
|
||||
pub(crate) mod h1;
|
||||
|
||||
pub(crate) use self::h1::{Conn, ServerTransaction};
|
||||
pub(crate) use self::h1::Conn;
|
||||
|
||||
#[cfg(feature = "client")]
|
||||
pub(crate) use self::h1::dispatch;
|
||||
#[cfg(feature = "server")]
|
||||
pub(crate) use self::h1::ServerTransaction;
|
||||
}
|
||||
|
||||
cfg_http2! {
|
||||
|
||||
@@ -49,6 +49,7 @@ pub(crate) use self::http::HttpService;
|
||||
#[cfg(feature = "client")]
|
||||
pub(crate) use self::make::MakeConnection;
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
#[cfg(feature = "server")]
|
||||
pub(crate) use self::make::MakeServiceRef;
|
||||
#[cfg(any(feature = "http1", feature = "http2"))]
|
||||
#[cfg(feature = "client")]
|
||||
|
||||
Reference in New Issue
Block a user