feat(server): introduce Accept trait

The `Accept` trait is used by the server types to asynchronously accept
incoming connections. This replaces the previous usage of `Stream`.

BREAKING CHANGE: Passing a `Stream` to `Server::builder` or
  `Http::serve_incoming` must be changed to pass an `Accept` instead. The
  `stream` optional feature can be enabled, and the a stream can be
  converted using `hyper::server:🉑:from_stream`.

Cargo.toml

@@ -69,6 +69,11 @@ runtime = [
     "tokio-net",
     "tokio-timer",
 ]
+
+# unstable features
+stream = []
+
+# internal features used in CI
 nightly = []
 __internal_flaky_tests = []
 __internal_happy_eyeballs_tests = []

src/server/accept.rs

@@ -0,0 +1,99 @@
+//! The `Accept` trait and supporting types.
+//!
+//! This module contains:
+//!
+//! - The [`Accept`](Accept) trait used to asynchronously accept incoming
+//!   connections.
+//! - Utilities like `poll_fn` to ease creating a custom `Accept`.
+
+#[cfg(feature = "stream")]
+use futures_core::Stream;
+
+use crate::common::{Pin, task::{self, Poll}};
+
+/// Asynchronously accept incoming connections.
+pub trait Accept {
+    /// The connection type that can be accepted.
+    type Conn;
+    /// The error type that can occur when accepting a connection.
+    type Error;
+
+    /// Poll to accept the next connection.
+    fn poll_accept(self: Pin<&mut Self>, cx: &mut task::Context<'_>)
+        -> Poll<Option<Result<Self::Conn, Self::Error>>>;
+}
+
+/// Create an `Accept` with a polling function.
+///
+/// # Example
+///
+/// ```
+/// use std::task::Poll;
+/// use hyper::server::{accept, Server};
+///
+/// # let mock_conn = ();
+/// // If we created some mocked connection...
+/// let mut conn = Some(mock_conn);
+///
+/// // And accept just the mocked conn once...
+/// let once = accept::poll_fn(move |cx| {
+///     Poll::Ready(conn.take().map(Ok::<_, ()>))
+/// });
+///
+/// let builder = Server::builder(once);
+/// ```
+pub fn poll_fn<F, IO, E>(func: F) -> impl Accept<Conn = IO, Error = E>
+where
+    F: FnMut(&mut task::Context<'_>) -> Poll<Option<Result<IO, E>>>,
+{
+    struct PollFn<F>(F);
+
+    impl<F, IO, E> Accept for PollFn<F>
+    where
+        F: FnMut(&mut task::Context<'_>) -> Poll<Option<Result<IO, E>>>,
+    {
+        type Conn = IO;
+        type Error = E;
+        fn poll_accept(self: Pin<&mut Self>, cx: &mut task::Context<'_>)
+            -> Poll<Option<Result<Self::Conn, Self::Error>>>
+        {
+            unsafe {
+                (self.get_unchecked_mut().0)(cx)
+            }
+        }
+    }
+
+    PollFn(func)
+}
+
+/// Adapt a `Stream` of incoming connections into an `Accept`.
+///
+/// # Unstable
+///
+/// This function requires enabling the unstable `stream` feature in your
+/// `Cargo.toml`.
+#[cfg(feature = "stream")]
+pub fn from_stream<S, IO, E>(stream: S) -> impl Accept<Conn = IO, Error = E>
+where
+    S: Stream<Item = Result<IO, E>>,
+{
+    struct FromStream<S>(S);
+
+    impl<S, IO, E> Accept for FromStream<S>
+    where
+        S: Stream<Item = Result<IO, E>>,
+    {
+        type Conn = IO;
+        type Error = E;
+        fn poll_accept(self: Pin<&mut Self>, cx: &mut task::Context<'_>)
+            -> Poll<Option<Result<Self::Conn, Self::Error>>>
+        {
+            unsafe {
+                Pin::new_unchecked(&mut self.get_unchecked_mut().0)
+                    .poll_next(cx)
+            }
+        }
+    }
+
+    FromStream(stream)
+}

src/server/conn.rs

@@ -28,6 +28,7 @@ use crate::error::{Kind, Parse};
 use crate::proto;
 use crate::service::{MakeServiceRef, Service};
 use crate::upgrade::Upgraded;
+use super::Accept;
 
 pub(super) use self::spawn_all::NoopWatcher;
 use self::spawn_all::NewSvcTask;
@@ -403,13 +404,10 @@ impl<E> Http<E> {
         }
     }
 
-    /// Bind the provided `addr` with the default `Handle` and return [`Serve`](Serve).
-    ///
-    /// This method will bind the `addr` provided with a new TCP listener ready
-    /// to accept connections. Each connection will be processed with the
-    /// `make_service` object provided, creating a new service per
-    /// connection.
     #[cfg(feature = "runtime")]
+    #[doc(hidden)]
+    #[deprecated]
+    #[allow(deprecated)]
     pub fn serve_addr<S, Bd>(&self, addr: &SocketAddr, make_service: S) -> crate::Result<Serve<AddrIncoming, S, E>>
     where
         S: MakeServiceRef<
@@ -428,13 +426,10 @@ impl<E> Http<E> {
         Ok(self.serve_incoming(incoming, make_service))
     }
 
-    /// Bind the provided `addr` with the `Handle` and return a [`Serve`](Serve)
-    ///
-    /// This method will bind the `addr` provided with a new TCP listener ready
-    /// to accept connections. Each connection will be processed with the
-    /// `make_service` object provided, creating a new service per
-    /// connection.
     #[cfg(feature = "runtime")]
+    #[doc(hidden)]
+    #[deprecated]
+    #[allow(deprecated)]
     pub fn serve_addr_handle<S, Bd>(&self, addr: &SocketAddr, handle: &Handle, make_service: S) -> crate::Result<Serve<AddrIncoming, S, E>>
     where
         S: MakeServiceRef<
@@ -453,10 +448,11 @@ impl<E> Http<E> {
         Ok(self.serve_incoming(incoming, make_service))
     }
 
-    /// Bind the provided stream of incoming IO objects with a `MakeService`.
+    #[doc(hidden)]
+    #[deprecated]
     pub fn serve_incoming<I, IO, IE, S, Bd>(&self, incoming: I, make_service: S) -> Serve<I, S, E>
     where
-        I: Stream<Item = Result<IO, IE>>,
+        I: Accept<Conn=IO, Error=IE>,
         IE: Into<Box<dyn StdError + Send + Sync>>,
         IO: AsyncRead + AsyncWrite + Unpin,
         S: MakeServiceRef<
@@ -678,13 +674,6 @@ where
 // ===== impl Serve =====
 
 impl<I, S, E> Serve<I, S, E> {
-    /// Spawn all incoming connections onto the executor in `Http`.
-    pub(super) fn spawn_all(self) -> SpawnAll<I, S, E> {
-        SpawnAll {
-            serve: self,
-        }
-    }
-
     /// Get a reference to the incoming stream.
     #[inline]
     pub fn incoming_ref(&self) -> &I {
@@ -696,22 +685,28 @@ impl<I, S, E> Serve<I, S, E> {
     pub fn incoming_mut(&mut self) -> &mut I {
         &mut self.incoming
     }
+
+    /// Spawn all incoming connections onto the executor in `Http`.
+    pub(super) fn spawn_all(self) -> SpawnAll<I, S, E> {
+        SpawnAll {
+            serve: self,
+        }
+    }
 }
 
-impl<I, IO, IE, S, B, E> Stream for Serve<I, S, E>
+
+
+
+impl<I, IO, IE, S, B, E> Serve<I, S, E>
 where
-    I: Stream<Item = Result<IO, IE>>,
+    I: Accept<Conn=IO, Error=IE>,
     IO: AsyncRead + AsyncWrite + Unpin,
     IE: Into<Box<dyn StdError + Send + Sync>>,
     S: MakeServiceRef<IO, Body, ResBody=B>,
-    //S::Error2: Into<Box<StdError + Send + Sync>>,
-    //SME: Into<Box<StdError + Send + Sync>>,
     B: Payload,
     E: H2Exec<<S::Service as Service<Body>>::Future, B>,
 {
-    type Item = crate::Result<Connecting<IO, S::Future, E>>;
+    fn poll_next_(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<crate::Result<Connecting<IO, S::Future, E>>>> {
-
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
         match ready!(self.project().make_service.poll_ready_ref(cx)) {
             Ok(()) => (),
             Err(e) => {
@@ -720,7 +715,7 @@ where
             }
         }
 
-        if let Some(item) = ready!(self.project().incoming.poll_next(cx)) {
+        if let Some(item) = ready!(self.project().incoming.poll_accept(cx)) {
             let io = item.map_err(crate::Error::new_accept)?;
             let new_fut = self.project().make_service.make_service_ref(&io);
             Poll::Ready(Some(Ok(Connecting {
@@ -734,6 +729,23 @@ where
     }
 }
 
+// deprecated
+impl<I, IO, IE, S, B, E> Stream for Serve<I, S, E>
+where
+    I: Accept<Conn=IO, Error=IE>,
+    IO: AsyncRead + AsyncWrite + Unpin,
+    IE: Into<Box<dyn StdError + Send + Sync>>,
+    S: MakeServiceRef<IO, Body, ResBody=B>,
+    B: Payload,
+    E: H2Exec<<S::Service as Service<Body>>::Future, B>,
+{
+    type Item = crate::Result<Connecting<IO, S::Future, E>>;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
+        self.poll_next_(cx)
+    }
+}
+
 // ===== impl Connecting =====
 
 
@@ -772,7 +784,7 @@ impl<I, S, E> SpawnAll<I, S, E> {
 
 impl<I, IO, IE, S, B, E> SpawnAll<I, S, E>
 where
-    I: Stream<Item=Result<IO, IE>>,
+    I: Accept<Conn=IO, Error=IE>,
     IE: Into<Box<dyn StdError + Send + Sync>>,
     IO: AsyncRead + AsyncWrite + Unpin + Send + 'static,
     S: MakeServiceRef<
@@ -789,7 +801,7 @@ where
         W: Watcher<IO, S::Service, E>,
     {
         loop {
-            if let Some(connecting) = ready!(self.project().serve.poll_next(cx)?) {
+            if let Some(connecting) = ready!(self.project().serve.poll_next_(cx)?) {
                 let fut = NewSvcTask::new(connecting, watcher.clone());
                 self.project().serve.project().protocol.exec.execute_new_svc(fut)?;
             } else {

src/server/mod.rs

@@ -48,6 +48,7 @@
 //! # fn main() {}
 //! ```
 
+pub mod accept;
 pub mod conn;
 mod shutdown;
 #[cfg(feature = "runtime")] mod tcp;
@@ -58,7 +59,6 @@ use std::fmt;
 
 #[cfg(feature = "runtime")] use std::time::Duration;
 
-use futures_core::Stream;
 use tokio_io::{AsyncRead, AsyncWrite};
 use pin_project::pin_project;
 
@@ -66,6 +66,7 @@ use crate::body::{Body, Payload};
 use crate::common::exec::{Exec, H2Exec, NewSvcExec};
 use crate::common::{Future, Pin, Poll, Unpin, task};
 use crate::service::{MakeServiceRef, Service};
+use self::accept::Accept;
 // Renamed `Http` as `Http_` for now so that people upgrading don't see an
 // error that `hyper::server::Http` is private...
 use self::conn::{Http as Http_, NoopWatcher, SpawnAll};
@@ -143,7 +144,7 @@ impl<S> Server<AddrIncoming, S> {
 
 impl<I, IO, IE, S, E, B> Server<I, S, E>
 where
-    I: Stream<Item=Result<IO, IE>>,
+    I: Accept<Conn=IO, Error=IE>,
     IE: Into<Box<dyn StdError + Send + Sync>>,
     IO: AsyncRead + AsyncWrite + Unpin + Send + 'static,
     S: MakeServiceRef<IO, Body, ResBody=B>,
@@ -200,7 +201,7 @@ where
 
 impl<I, IO, IE, S, B, E> Future for Server<I, S, E>
 where
-    I: Stream<Item=Result<IO, IE>>,
+    I: Accept<Conn=IO, Error=IE>,
     IE: Into<Box<dyn StdError + Send + Sync>>,
     IO: AsyncRead + AsyncWrite + Unpin + Send + 'static,
     S: MakeServiceRef<IO, Body, ResBody=B>,
@@ -380,17 +381,17 @@ impl<I, E> Builder<I, E> {
     /// // Finally, spawn `server` onto an Executor...
     /// # }
     /// ```
-    pub fn serve<S, B, IO, IE>(self, new_service: S) -> Server<I, S, E>
+    pub fn serve<S, B>(self, new_service: S) -> Server<I, S, E>
     where
-        I: Stream<Item=Result<IO, IE>>,
+        I: Accept,
-        IE: Into<Box<dyn StdError + Send + Sync>>,
+        I::Error: Into<Box<dyn StdError + Send + Sync>>,
-        IO: AsyncRead + AsyncWrite + Unpin + Send + 'static,
+        I::Conn: AsyncRead + AsyncWrite + Unpin + Send + 'static,
-        S: MakeServiceRef<IO, Body, ResBody=B>,
+        S: MakeServiceRef<I::Conn, Body, ResBody=B>,
         S::Error: Into<Box<dyn StdError + Send + Sync>>,
         S::Service: 'static,
         B: Payload,
         B::Data: Unpin,
-        E: NewSvcExec<IO, S::Future, S::Service, E, NoopWatcher>,
+        E: NewSvcExec<I::Conn, S::Future, S::Service, E, NoopWatcher>,
         E: H2Exec<<S::Service as Service<Body>>::Future, B>,
     {
         let serve = self.protocol.serve_incoming(self.incoming, new_service);

src/server/shutdown.rs

@@ -1,6 +1,5 @@
 use std::error::Error as StdError;
 
-use futures_core::Stream;
 use tokio_io::{AsyncRead, AsyncWrite};
 use pin_project::{pin_project, project};
 
@@ -9,6 +8,7 @@ use crate::common::drain::{self, Draining, Signal, Watch, Watching};
 use crate::common::exec::{H2Exec, NewSvcExec};
 use crate::common::{Future, Pin, Poll, Unpin, task};
 use crate::service::{MakeServiceRef, Service};
+use super::Accept;
 use super::conn::{SpawnAll, UpgradeableConnection, Watcher};
 
 #[allow(missing_debug_implementations)]
@@ -46,7 +46,7 @@ impl<I, S, F, E> Graceful<I, S, F, E> {
 
 impl<I, IO, IE, S, B, F, E> Future for Graceful<I, S, F, E>
 where
-    I: Stream<Item=Result<IO, IE>>,
+    I: Accept<Conn=IO, Error=IE>,
     IE: Into<Box<dyn StdError + Send + Sync>>,
     IO: AsyncRead + AsyncWrite + Unpin + Send + 'static,
     S: MakeServiceRef<IO, Body, ResBody=B>,

src/server/tcp.rs

@@ -3,7 +3,6 @@ use std::io;
 use std::net::{SocketAddr, TcpListener as StdTcpListener};
 use std::time::Duration;
 
-use futures_core::Stream;
 use futures_util::FutureExt as _;
 use tokio_net::driver::Handle;
 use tokio_net::tcp::TcpListener;
@@ -11,6 +10,7 @@ use tokio_timer::Delay;
 
 use crate::common::{Future, Pin, Poll, task};
 
+use super::Accept;
 pub use self::addr_stream::AddrStream;
 
 /// A stream of connections from binding to an address.
@@ -156,6 +156,7 @@ impl AddrIncoming {
     }
 }
 
+/*
 impl Stream for AddrIncoming {
     type Item = io::Result<AddrStream>;
 
@@ -164,6 +165,17 @@ impl Stream for AddrIncoming {
         Poll::Ready(Some(result))
     }
 }
+*/
+
+impl Accept for AddrIncoming {
+    type Conn = AddrStream;
+    type Error = io::Error;
+
+    fn poll_accept(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Result<Self::Conn, Self::Error>>> {
+        let result = ready!(self.poll_next_(cx));
+        Poll::Ready(Some(result))
+    }
+}
 
 /// This function defines errors that are per-connection. Which basically
 /// means that if we get this error from `accept()` system call it means

tests/support/mod.rs

@@ -322,7 +322,18 @@ pub fn __run_test(cfg: __TestConfig) {
     let serve_handles = Arc::new(Mutex::new(
         cfg.server_msgs
     ));
+
+    let expected_connections = cfg.connections;
+    let mut cnt = 0;
     let new_service = make_service_fn(move |_| {
+        cnt += 1;
+        assert!(
+            cnt <= expected_connections,
+            "server expected {} connections, received {}",
+            expected_connections,
+            cnt
+        );
+
         // Move a clone into the service_fn
         let serve_handles = serve_handles.clone();
         future::ok::<_, hyper::Error>(service_fn(move |req: Request<Body>| {
@@ -352,36 +363,15 @@ pub fn __run_test(cfg: __TestConfig) {
         }))
     });
 
-    let serve = hyper::server::conn::Http::new()
+    let server = hyper::Server::bind(&SocketAddr::from(([127, 0, 0, 1], 0)))
         .http2_only(cfg.server_version == 2)
-        .serve_addr(
+        .serve(new_service);
-            &SocketAddr::from(([127, 0, 0, 1], 0)),
-            new_service,
-        )
-        .expect("serve_addr");
 
-    let mut addr = serve.incoming_ref().local_addr();
+    let mut addr = server.local_addr();
-    let expected_connections = cfg.connections;
-    let server = serve
-        .try_fold(0, move |cnt, connecting| {
-            let cnt = cnt + 1;
-            assert!(
-                cnt <= expected_connections,
-                "server expected {} connections, received {}",
-                expected_connections,
-                cnt
-            );
-            let fut = connecting
-                .then(|res| res.expect("connecting"))
-                .map(|conn_res| conn_res.expect("server connection error"));
-            tokio::spawn(fut);
-            future::ok::<_, hyper::Error>(cnt)
-        })
-        .map(|res| {
-            let _ = res.expect("serve error");
-        });
 
-    rt.spawn(server);
+    rt.spawn(server.map(|result| {
+        let _ = result.expect("server error");
+    }));
 
     if cfg.proxy {
         let (proxy_addr, proxy) = naive_proxy(ProxyConfig {
@@ -393,7 +383,6 @@ pub fn __run_test(cfg: __TestConfig) {
         addr = proxy_addr;
     }
 
-
     let make_request = Arc::new(move |client: &Client<HttpConnector>, creq: __CReq, cres: __CRes| {
         let uri = format!("http://{}{}", addr, creq.uri);
         let mut req = Request::builder()