refactor: no more futures crate

Cargo.toml

@@ -18,7 +18,9 @@ rust-version = "1.65"
 bincode = { version = "1.3.3", optional = true }
 bytes = { version = "1", optional = true }
 flume = { version = "0.11", optional = true }
-futures = "0.3"
+futures-lite = "2.3.0"
+futures-sink = "0.3.30"
+futures-util = { version = "0.3.30", features = ["sink"] }
 hyper = { version = "0.14.16", features = ["full"], optional = true }
 pin-project = "1"
 quinn = { version = "0.10", optional = true }
@@ -47,6 +49,7 @@ thousands = "0.2.0"
 tracing-subscriber = "0.3.16"
 tempfile = "3.5.0"
 proc-macro2 = "1.0.66"
+futures-buffered = "0.2.4"
 
 [features]
 hyper-transport = ["flume", "hyper", "bincode", "bytes"]
@@ -77,4 +80,3 @@ required-features = ["flume-transport"]
 
 [workspace]
 members = ["examples/split/types", "examples/split/server", "examples/split/client"]
-

examples/macro.rs

@@ -49,7 +49,8 @@ mod store_rpc {
 }
 
 use async_stream::stream;
-use futures::{SinkExt, Stream, StreamExt};
+use futures_lite::{Stream, StreamExt};
+use futures_util::SinkExt;
 use quic_rpc::client::RpcClient;
 use quic_rpc::server::run_server_loop;
 use quic_rpc::transport::flume;

examples/modularize.rs

@@ -8,7 +8,8 @@
 //! unchanged.
 
 use anyhow::Result;
-use futures::{SinkExt, TryStreamExt};
+use futures_lite::StreamExt;
+use futures_util::SinkExt;
 use quic_rpc::{transport::flume, RpcClient, RpcServer, ServiceConnection, ServiceEndpoint};
 use tracing::warn;
 
@@ -272,7 +273,7 @@ mod calc {
 
     use anyhow::{bail, Result};
     use derive_more::{From, TryInto};
-    use futures::{Stream, StreamExt};
+    use futures_lite::{Stream, StreamExt};
     use quic_rpc::{
         message::{ClientStreaming, ClientStreamingMsg, Msg, RpcMsg},
         server::RpcChannel,
@@ -394,7 +395,8 @@ mod clock {
 
     use anyhow::Result;
     use derive_more::{From, TryInto};
-    use futures::{stream::BoxStream, Stream, StreamExt, TryStreamExt};
+    use futures_lite::{stream::Boxed as BoxStream, Stream, StreamExt};
+    use futures_util::TryStreamExt;
     use quic_rpc::{
         message::{Msg, ServerStreaming, ServerStreamingMsg},
         server::RpcChannel,
@@ -524,7 +526,7 @@ mod clock {
         pub fn new(client: RpcClient<S, C, ClockService>) -> Self {
             Self { client }
         }
-        pub async fn tick(&self) -> Result<BoxStream<'static, Result<usize>>> {
+        pub async fn tick(&self) -> Result<BoxStream<Result<usize>>> {
             let res = self.client.server_streaming(TickRequest).await?;
             Ok(res.map_ok(|r| r.tick).map_err(anyhow::Error::from).boxed())
         }

examples/store.rs

@@ -1,7 +1,8 @@
 #![allow(clippy::enum_variant_names)]
 use async_stream::stream;
 use derive_more::{From, TryInto};
-use futures::{SinkExt, Stream, StreamExt};
+use futures_lite::{Stream, StreamExt};
+use futures_util::SinkExt;
 use quic_rpc::{
     server::RpcServerError,
     transport::{flume, Connection, ServerEndpoint},

src/client.rs

@@ -5,7 +5,9 @@ use crate::{
     map::{ChainedMapper, MapService, Mapper},
     Service, ServiceConnection,
 };
-use futures::{Sink, SinkExt, Stream};
+use futures_lite::Stream;
+use futures_sink::Sink;
+
 use pin_project::pin_project;
 use std::{
     fmt::Debug,
@@ -62,20 +64,20 @@ where
     type Error = C::SendError;
 
     fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.project().0.poll_ready_unpin(cx)
+        self.project().0.poll_ready(cx)
     }
 
     fn start_send(self: Pin<&mut Self>, item: T) -> Result<(), Self::Error> {
         let req = self.2.req_into_outer(item.into());
-        self.project().0.start_send_unpin(req)
+        self.project().0.start_send(req)
     }
 
     fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.project().0.poll_flush_unpin(cx)
+        self.project().0.poll_flush(cx)
     }
 
     fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.project().0.poll_close_unpin(cx)
+        self.project().0.poll_close(cx)
     }
 }
 

src/pattern/bidi_streaming.rs

@@ -1,5 +1,7 @@
 //!
-use futures::{FutureExt, SinkExt, Stream, StreamExt};
+
+use futures_lite::{Stream, StreamExt};
+use futures_util::{FutureExt, SinkExt};
 
 use crate::{
     client::{BoxStreamSync, UpdateSink},

src/pattern/client_streaming.rs

@@ -1,5 +1,6 @@
 //!
-use futures::{future::BoxFuture, Future, FutureExt, SinkExt, StreamExt, TryFutureExt};
+use futures_lite::{future::Boxed, Future, StreamExt};
+use futures_util::{FutureExt, SinkExt, TryFutureExt};
 
 use crate::{
     client::UpdateSink,
@@ -88,7 +89,7 @@ where
     ) -> result::Result<
         (
             UpdateSink<S, C, M::Update, SInner>,
-            BoxFuture<'static, result::Result<M::Response, ItemError<C>>>,
+            Boxed<result::Result<M::Response, ItemError<C>>>,
         ),
         Error<C>,
     >

src/pattern/rpc.rs

@@ -1,5 +1,7 @@
 //!
-use futures::{Future, FutureExt, SinkExt, StreamExt};
+
+use futures_lite::{Future, StreamExt};
+use futures_util::{FutureExt, SinkExt};
 
 use crate::{
     message::{InteractionPattern, Msg},

src/pattern/server_streaming.rs

@@ -1,5 +1,7 @@
 //!
-use futures::{FutureExt, SinkExt, Stream, StreamExt, TryFutureExt};
+
+use futures_lite::{Stream, StreamExt};
+use futures_util::{FutureExt, SinkExt, TryFutureExt};
 
 use crate::{
     client::{BoxStreamSync, DeferDrop},

src/pattern/try_server_streaming.rs

@@ -1,5 +1,6 @@
 //!
-use futures::{FutureExt, SinkExt, Stream, StreamExt, TryFutureExt};
+use futures_lite::{Future, Stream, StreamExt};
+use futures_util::{FutureExt, SinkExt, TryFutureExt};
 use serde::{Deserialize, Serialize};
 
 use crate::{
@@ -120,7 +121,7 @@ where
         std::result::Result<StreamCreated, M::CreateError>:
             Into<SInner::Res> + TryFrom<SInner::Res>,
         F: FnOnce(T, M) -> Fut + Send + 'static,
-        Fut: futures::Future<Output = std::result::Result<Str, M::CreateError>> + Send + 'static,
+        Fut: Future<Output = std::result::Result<Str, M::CreateError>> + Send + 'static,
         Str: Stream<Item = std::result::Result<M::Item, M::ItemError>> + Send + 'static,
         T: Send + 'static,
     {

src/server.rs

@@ -6,7 +6,7 @@ use crate::{
     transport::ConnectionErrors,
     Service, ServiceEndpoint,
 };
-use futures::{channel::oneshot, task, task::Poll, Future, FutureExt, Stream, StreamExt};
+use futures_lite::{Future, Stream, StreamExt};
 use pin_project::pin_project;
 use std::{
     error,
@@ -15,7 +15,9 @@ use std::{
     pin::Pin,
     result,
     sync::Arc,
+    task::{self, Poll},
 };
+use tokio::sync::oneshot;
 
 /// A server channel for a specific service.
 ///
@@ -200,7 +202,7 @@ where
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
         let mut this = self.project();
-        match this.0.poll_next_unpin(cx) {
+        match Pin::new(&mut this.0).poll_next(cx) {
             Poll::Ready(Some(msg)) => match msg {
                 Ok(msg) => {
                     let msg = this.3.req_try_into_inner(msg);
@@ -274,7 +276,7 @@ impl<T> Future for UnwrapToPending<T> {
     type Output = T;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
-        match self.0.poll_unpin(cx) {
+        match Pin::new(&mut self.0).poll(cx) {
             Poll::Ready(Ok(x)) => Poll::Ready(x),
             Poll::Ready(Err(_)) => Poll::Pending,
             Poll::Pending => Poll::Pending,

src/transport/combined.rs

@@ -1,10 +1,9 @@
 //! Transport that combines two other transports
 use super::{Connection, ConnectionCommon, ConnectionErrors, LocalAddr, ServerEndpoint};
 use crate::RpcMessage;
-use futures::{
-    future::{self, BoxFuture},
-    FutureExt, Sink, Stream, TryFutureExt,
-};
+use futures_lite::{future::Boxed as BoxFuture, Stream};
+use futures_sink::Sink;
+use futures_util::{FutureExt, TryFutureExt};
 use pin_project::pin_project;
 use std::{
     error, fmt,
@@ -284,11 +283,11 @@ impl<A: ConnectionErrors, B: ConnectionErrors> error::Error for AcceptBiError<A,
 
 /// Future returned by open_bi
 pub type OpenBiFuture<A, B, In, Out> =
-    BoxFuture<'static, result::Result<Socket<A, B, In, Out>, self::OpenBiError<A, B>>>;
+    BoxFuture<result::Result<Socket<A, B, In, Out>, self::OpenBiError<A, B>>>;
 
 /// Future returned by accept_bi
 pub type AcceptBiFuture<A, B, In, Out> =
-    BoxFuture<'static, result::Result<self::Socket<A, B, In, Out>, self::AcceptBiError<A, B>>>;
+    BoxFuture<result::Result<self::Socket<A, B, In, Out>, self::AcceptBiError<A, B>>>;
 
 type Socket<A, B, In, Out> = (
     self::SendSink<A, B, In, Out>,
@@ -324,7 +323,7 @@ impl<A: Connection<In, Out>, B: Connection<In, Out>, In: RpcMessage, Out: RpcMes
                 let (send, recv) = b.open_bi().await.map_err(OpenBiError::B)?;
                 Ok((SendSink::B(send), RecvStream::B(recv)))
             } else {
-                future::err(OpenBiError::NoChannel).await
+                std::future::ready(Err(OpenBiError::NoChannel)).await
             }
         }
         .boxed()
@@ -363,7 +362,7 @@ impl<A: ServerEndpoint<In, Out>, B: ServerEndpoint<In, Out>, In: RpcMessage, Out
                 .map_err(AcceptBiError::A)
                 .left_future()
         } else {
-            future::pending().right_future()
+            std::future::pending().right_future()
         };
         let b_fut = if let Some(b) = &self.b {
             b.accept_bi()
@@ -376,7 +375,7 @@ impl<A: ServerEndpoint<In, Out>, B: ServerEndpoint<In, Out>, In: RpcMessage, Out
                 .map_err(AcceptBiError::B)
                 .left_future()
         } else {
-            future::pending().right_future()
+            std::future::pending().right_future()
         };
         async move {
             tokio::select! {