Upgrade to Tokio 1.0

.github/workflows/ci.yml

@@ -45,8 +45,8 @@ jobs:
             benches: true
           - build: tls
             features: "--features tls"
-          - build: uds
-            features: "--features tokio/uds"
+          - build: net
+            features: "--features tokio/net"
           - build: no-default-features
             features: "--no-default-features"
           - build: compression

Cargo.toml

@@ -17,12 +17,12 @@ edition = "2018"
 all-features = true
 
 [dependencies]
-async-compression = { version = "0.3.1", features = ["brotli", "deflate", "gzip", "stream"], optional = true }
-bytes = "0.5"
+async-compression = { version = "0.3.7", features = ["brotli", "deflate", "gzip", "tokio"], optional = true }
+bytes = "1.0"
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
 headers = "0.3"
 http = "0.2"
-hyper = { version = "0.13", features = ["stream"] }
+hyper = { version = "0.14", features = ["stream", "server", "http1", "http2", "tcp", "client"] }
 log = "0.4"
 mime = "0.3"
 mime_guess = "2.0.0"
@@ -31,23 +31,25 @@ scoped-tls = "1.0"
 serde = "1.0"
 serde_json = "1.0"
 serde_urlencoded = "0.7"
-tokio = { version = "0.2", features = ["fs", "stream", "sync", "time"] }
+tokio = { version = "1.0", features = ["fs", "sync", "time"] }
+tokio-util = { version = "0.6", features = ["io"] }
 tracing = { version = "0.1", default-features = false, features = ["log", "std"] }
 tracing-futures = { version = "0.2", default-features = false, features = ["std-future"] }
 tower-service = "0.3"
 # tls is enabled by default, we don't want that yet
-tokio-tungstenite = { version = "0.11", default-features = false, optional = true }
+tokio-tungstenite = { git = "https://github.com/aknuds1/tokio-tungstenite.git", rev = "cad06f3ac07e7fc2a936308207aab72b31478462", default-features = false, optional = true }
 percent-encoding = "2.1"
 pin-project = "1.0"
-tokio-rustls = { version = "0.14", optional = true }
+tokio-rustls = { version = "0.22", optional = true }
+async-stream = "0.3"
 
 [dev-dependencies]
 pretty_env_logger = "0.4"
 tracing-subscriber = "0.2.7"
 tracing-log = "0.1"
 serde_derive = "1.0"
 handlebars = "3.0.0"
-tokio = { version = "0.2", features = ["macros"] }
+tokio = { version = "1.0", features = ["macros", "rt-multi-thread"] }
 listenfd = "0.3"
 
 [features]
@@ -78,7 +80,6 @@ required-features = ["compression"]
 
 [[example]]
 name = "unix_socket"
-required-features = ["tokio/uds"]
 
 [[example]]
 name = "websockets"

examples/futures.rs

@@ -16,7 +16,7 @@ async fn main() {
 }
 
 async fn sleepy(Seconds(seconds): Seconds) -> Result<impl warp::Reply, Infallible> {
-    tokio::time::delay_for(Duration::from_secs(seconds)).await;
+    tokio::time::sleep(Duration::from_secs(seconds)).await;
     Ok(format!("I waited {} seconds!", seconds))
 }
 

examples/sse.rs

@@ -1,3 +1,4 @@
+use async_stream::stream;
 use futures::StreamExt;
 use std::convert::Infallible;
 use std::time::Duration;
@@ -16,7 +17,13 @@ async fn main() {
     let routes = warp::path("ticks").and(warp::get()).map(|| {
         let mut counter: u64 = 0;
         // create server event source
-        let event_stream = interval(Duration::from_secs(1)).map(move |_| {
+        let mut interval = interval(Duration::from_secs(1));
+        let stream = stream! {
+            while let item = interval.tick().await {
+                yield item;
+            }
+        };
+        let event_stream = stream.map(move |_| {
             counter += 1;
             sse_counter(counter)
         });

examples/sse_chat.rs

@@ -5,6 +5,7 @@ use std::sync::{
     Arc, Mutex,
 };
 use tokio::sync::mpsc;
+use warp::test;
 use warp::{sse::ServerSentEvent, Filter};
 
 #[tokio::main]
@@ -85,7 +86,7 @@ fn user_connected(
 
     // Use an unbounded channel to handle buffering and flushing of messages
     // to the event source...
-    let (tx, rx) = mpsc::unbounded_channel();
+    let (tx, rx) = test::unbounded_channel_stream();
 
     tx.send(Message::UserId(my_id))
         // rx is right above, so this cannot fail

examples/unix_socket.rs

@@ -1,13 +1,18 @@
 #![deny(warnings)]
 
+use async_stream::stream;
 use tokio::net::UnixListener;
 
 #[tokio::main]
 async fn main() {
     pretty_env_logger::init();
 
-    let mut listener = UnixListener::bind("/tmp/warp.sock").unwrap();
-    let incoming = listener.incoming();
+    let listener = UnixListener::bind("/tmp/warp.sock").unwrap();
+    let incoming = stream! {
+        while let item = listener.accept().await {
+            yield item.map(|item| item.0);
+        }
+    };
     warp::serve(warp::fs::dir("examples/dir"))
         .run_incoming(incoming)
         .await;

examples/websockets_chat.rs

@@ -7,6 +7,7 @@ use std::sync::{
 
 use futures::{FutureExt, StreamExt};
 use tokio::sync::{mpsc, RwLock};
+use warp::test;
 use warp::ws::{Message, WebSocket};
 use warp::Filter;
 
@@ -58,7 +59,7 @@ async fn user_connected(ws: WebSocket, users: Users) {
 
     // Use an unbounded channel to handle buffering and flushing of messages
     // to the websocket...
-    let (tx, rx) = mpsc::unbounded_channel();
+    let (tx, rx) = test::unbounded_channel_stream();
     tokio::task::spawn(rx.forward(user_ws_tx).map(|result| {
         if let Err(e) = result {
             eprintln!("websocket send error: {}", e);

src/filters/body.rs

@@ -7,7 +7,7 @@ use std::fmt;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
-use bytes::{buf::BufExt, Buf, Bytes};
+use bytes::{Buf, Bytes};
 use futures::{future, ready, Stream, TryFutureExt};
 use headers::ContentLength;
 use http::header::CONTENT_TYPE;
@@ -131,8 +131,8 @@ pub fn bytes() -> impl Filter<Extract = (Bytes,), Error = Rejection> + Copy {
 /// fn full_body(mut body: impl Buf) {
 ///     // It could have several non-contiguous slices of memory...
 ///     while body.has_remaining() {
-///         println!("slice = {:?}", body.bytes());
-///         let cnt = body.bytes().len();
+///         println!("slice = {:?}", body.chunk());
+///         let cnt = body.chunk().len();
 ///         body.advance(cnt);
 ///     }
 /// }
@@ -232,7 +232,7 @@ impl Decode for Json {
     const WITH_NO_CONTENT_TYPE: bool = true;
 
     fn decode<B: Buf, T: DeserializeOwned>(mut buf: B) -> Result<T, BoxError> {
-        serde_json::from_slice(&buf.to_bytes()).map_err(Into::into)
+        serde_json::from_slice(&buf.copy_to_bytes(buf.remaining())).map_err(Into::into)
     }
 }
 

src/filters/compression.rs

@@ -2,12 +2,14 @@
 //!
 //! Filters that compress the body of a response.
 
-use async_compression::stream::{BrotliEncoder, DeflateEncoder, GzipEncoder};
+use async_compression::tokio::bufread::{BrotliEncoder, DeflateEncoder, GzipEncoder};
 use http::header::HeaderValue;
 use hyper::{
     header::{CONTENT_ENCODING, CONTENT_LENGTH},
     Body,
 };
+use tokio_util::codec::{BytesCodec, FramedRead};
+use tokio_util::io::StreamReader;
 
 use crate::filter::{Filter, WrapSealed};
 use crate::reject::IsReject;
@@ -56,7 +58,10 @@ pub struct Compression<F> {
 /// ```
 pub fn gzip() -> Compression<impl Fn(CompressionProps) -> Response + Copy> {
     let func = move |mut props: CompressionProps| {
-        let body = Body::wrap_stream(GzipEncoder::new(props.body));
+        let body = Body::wrap_stream(FramedRead::new(
+            GzipEncoder::new(StreamReader::new(props.body)),
+            BytesCodec::new(),
+        ));
         props
             .head
             .headers
@@ -82,7 +87,10 @@ pub fn gzip() -> Compression<impl Fn(CompressionProps) -> Response + Copy> {
 /// ```
 pub fn deflate() -> Compression<impl Fn(CompressionProps) -> Response + Copy> {
     let func = move |mut props: CompressionProps| {
-        let body = Body::wrap_stream(DeflateEncoder::new(props.body));
+        let body = Body::wrap_stream(FramedRead::new(
+            DeflateEncoder::new(StreamReader::new(props.body)),
+            BytesCodec::new(),
+        ));
         props
             .head
             .headers
@@ -108,7 +116,10 @@ pub fn deflate() -> Compression<impl Fn(CompressionProps) -> Response + Copy> {
 /// ```
 pub fn brotli() -> Compression<impl Fn(CompressionProps) -> Response + Copy> {
     let func = move |mut props: CompressionProps| {
-        let body = Body::wrap_stream(BrotliEncoder::new(props.body));
+        let body = Body::wrap_stream(FramedRead::new(
+            BrotliEncoder::new(StreamReader::new(props.body)),
+            BytesCodec::new(),
+        ));
         props
             .head
             .headers

src/filters/fs.rs

@@ -22,7 +22,8 @@ use hyper::Body;
 use mime_guess;
 use percent_encoding::percent_decode_str;
 use tokio::fs::File as TkFile;
-use tokio::io::AsyncRead;
+use tokio::io::AsyncSeekExt;
+use tokio_util::io::poll_read_buf;
 
 use crate::filter::{Filter, FilterClone, One};
 use crate::reject::{self, Rejection};
@@ -419,7 +420,7 @@ fn file_stream(
                 }
                 reserve_at_least(&mut buf, buf_size);
 
-                let n = match ready!(Pin::new(&mut f).poll_read_buf(cx, &mut buf)) {
+                let n = match ready!(poll_read_buf(Pin::new(&mut f), cx, &mut buf)) {
                     Ok(n) => n as u64,
                     Err(err) => {
                         tracing::debug!("file read error: {}", err);

src/filters/sse.rs

@@ -54,7 +54,7 @@ use hyper::Body;
 use pin_project::pin_project;
 use serde::Serialize;
 use serde_json;
-use tokio::time::{self, Delay};
+use tokio::time::{self, Sleep};
 
 use self::sealed::{
     BoxedServerSentEvent, EitherServerSentEvent, SseError, SseField, SseFormat, SseWrapper,
@@ -467,7 +467,7 @@ impl KeepAlive {
         S::Ok: ServerSentEvent + Send,
         S::Error: StdError + Send + Sync + 'static,
     {
-        let alive_timer = time::delay_for(self.max_interval);
+        let alive_timer = time::sleep(self.max_interval);
         SseKeepAlive {
             event_stream,
             comment_text: self.comment_text,
@@ -484,7 +484,8 @@ struct SseKeepAlive<S> {
     event_stream: S,
     comment_text: Cow<'static, str>,
     max_interval: Duration,
-    alive_timer: Delay,
+    #[pin]
+    alive_timer: Sleep,
 }
 
 #[doc(hidden)]
@@ -505,7 +506,7 @@ where
     let max_interval = keep_interval
         .into()
         .unwrap_or_else(|| Duration::from_secs(15));
-    let alive_timer = time::delay_for(max_interval);
+    let alive_timer = time::sleep(max_interval);
     SseKeepAlive {
         event_stream,
         comment_text: Cow::Borrowed(""),
@@ -530,6 +531,7 @@ where
 /// use futures::StreamExt;
 /// use tokio::time::interval;
 /// use warp::{Filter, Stream, sse::ServerSentEvent};
+/// use async_stream::stream;
 ///
 /// // create server-sent event
 /// fn sse_counter(counter: u64) ->  Result<impl ServerSentEvent, Infallible> {
@@ -541,7 +543,13 @@ where
 ///         .and(warp::get())
 ///         .map(|| {
 ///             let mut counter: u64 = 0;
-///             let event_stream = interval(Duration::from_secs(15)).map(move |_| {
+///             let mut interval = interval(Duration::from_secs(15));
+///             let stream = stream! {
+///                 while let item = interval.tick().await {
+///                     yield item;
+///                 }
+///             };
+///             let event_stream = stream.map(move |_| {
 ///                 counter += 1;
 ///                 sse_counter(counter)
 ///             });

src/filters/ws.rs

@@ -7,12 +7,13 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 
 use super::{body, header};
-use crate::filter::{Filter, One};
+use crate::filter::{filter_fn_one, Filter, One};
 use crate::reject::Rejection;
 use crate::reply::{Reply, Response};
 use futures::{future, ready, FutureExt, Sink, Stream, TryFutureExt};
 use headers::{Connection, HeaderMapExt, SecWebsocketAccept, SecWebsocketKey, Upgrade};
 use http;
+use hyper::upgrade::OnUpgrade;
 use tokio_tungstenite::{
     tungstenite::protocol::{self, WebSocketConfig},
     WebSocketStream,
@@ -58,18 +59,23 @@ pub fn ws() -> impl Filter<Extract = One<Ws>, Error = Rejection> + Copy {
         //.and(header::exact2(SecWebsocketVersion::V13))
         .and(header::header2::<SecWebsocketKey>())
         .and(body::body())
-        .map(move |key: SecWebsocketKey, body: ::hyper::Body| Ws {
-            body,
-            config: None,
-            key,
-        })
+        .and(on_upgrade())
+        .map(
+            move |key: SecWebsocketKey, body: ::hyper::Body, on_upgrade: Option<OnUpgrade>| Ws {
+                body,
+                config: None,
+                key,
+                on_upgrade,
+            },
+        )
 }
 
 /// Extracted by the [`ws`](ws) filter, and used to finish an upgrade.
 pub struct Ws {
     body: ::hyper::Body,
     config: Option<WebSocketConfig>,
     key: SecWebsocketKey,
+    on_upgrade: Option<OnUpgrade>,
 }
 
 impl Ws {
@@ -134,10 +140,11 @@ where
     fn into_response(self) -> Response {
         let on_upgrade = self.on_upgrade;
         let config = self.ws.config;
-        let fut = self
-            .ws
-            .body
-            .on_upgrade()
+        let mut resp = Response::new(self.ws.body);
+        if let Some(on_upgrade) = self.ws.on_upgrade {
+            resp.extensions_mut().insert(on_upgrade);
+        }
+        let fut = hyper::upgrade::on(resp)
             .and_then(move |upgraded| {
                 tracing::trace!("websocket upgrade complete");
                 WebSocket::from_raw_socket(upgraded, protocol::Role::Server, config).map(Ok)
@@ -163,6 +170,11 @@ where
     }
 }
 
+// Extracts OnUpgrade state from the route.
+fn on_upgrade() -> impl Filter<Extract = (Option<OnUpgrade>,), Error = Rejection> + Copy {
+    filter_fn_one(|route| future::ready(Ok(route.extensions_mut().remove::<OnUpgrade>())))
+}
+
 /// A websocket `Stream` and `Sink`, provided to `ws` filters.
 ///
 /// Ping messages sent from the client will be handled internally by replying with a Pong message.

src/route.rs

@@ -75,11 +75,10 @@ impl Route {
         self.req.extensions()
     }
 
-    /*
+    #[cfg(feature = "websocket")]
     pub(crate) fn extensions_mut(&mut self) -> &mut http::Extensions {
         self.req.extensions_mut()
     }
-    */
 
     pub(crate) fn uri(&self) -> &http::Uri {
         self.req.uri()