diff --git a/README.md b/README.md index a471632..1dd7dfc 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ To use with your project, add the following to your Cargo.toml: stream-reconnect = "0.3" ``` -*Minimum supported rust version: 1.43.1* +_Minimum supported rust version: 1.43.1_ ## Runtime Support @@ -54,12 +54,10 @@ impl UnderlyingStream, WsError> for MyWs { // Establishes connection. // Additionally, this will be used when reconnect tries are attempted. - fn establish(addr: String) -> Pin> + Send>> { - Box::pin(async move { - // In this case, we are trying to connect to the WebSocket endpoint - let ws_connection = connect_async(addr).await.unwrap().0; - Ok(ws_connection) - }) + async fn establish(addr: String) -> Result { + // In this case, we are trying to connect to the WebSocket endpoint + let ws_connection = connect_async(addr).await.unwrap().0; + Ok(ws_connection) } // The following errors are considered disconnect errors. @@ -97,4 +95,4 @@ ws_stream.send("hello world!".into()).await.unwrap(); ## License -MIT \ No newline at end of file +MIT diff --git a/src/stream.rs b/src/stream.rs index 2c6e7d2..789096c 100644 --- a/src/stream.rs +++ b/src/stream.rs @@ -7,7 +7,8 @@ use std::pin::Pin; use std::task::{Context, Poll}; use std::time::Duration; -use futures::{ready, Sink, Stream}; +use futures::future::BoxFuture; +use futures::{ready, FutureExt, Sink, Stream}; use log::{debug, error, info}; use crate::config::ReconnectOptions; @@ -24,12 +25,12 @@ where /// The creation function is used by [ReconnectStream] in order to establish both the initial IO connection /// in addition to performing reconnects. #[cfg(feature = "not-send")] - fn establish(ctor_arg: C) -> Pin>>>; + fn establish(ctor_arg: C) -> impl Future>; /// The creation function is used by [ReconnectStream] in order to establish both the initial IO connection /// in addition to performing reconnects. #[cfg(not(feature = "not-send"))] - fn establish(ctor_arg: C) -> Pin> + Send>>; + fn establish(ctor_arg: C) -> impl Future> + Send; /// When sink send experience an `Error` during operation, it does not necessarily mean /// it is a disconnect/termination (ex: WouldBlock). @@ -62,9 +63,9 @@ where { attempts_tracker: AttemptsTracker, #[cfg(not(feature = "not-send"))] - reconnect_attempt: Pin> + Send>>, + reconnect_attempt: BoxFuture<'static, Result>, #[cfg(feature = "not-send")] - reconnect_attempt: Pin>>>, + reconnect_attempt: LocalBoxFuture<'static, Result>, _marker: PhantomData<(C, I, E)>, } @@ -80,7 +81,7 @@ where attempt_num: 0, retries_remaining: (options.retries_to_attempt_fn())(), }, - reconnect_attempt: Box::pin(async { unreachable!("Not going to happen") }), + reconnect_attempt: async { unreachable!("Not going to happen") }.boxed(), _marker: PhantomData, } } @@ -245,14 +246,12 @@ where reconnect_status.attempts_tracker.attempt_num += 1; let cur_num = reconnect_status.attempts_tracker.attempt_num; - - let reconnect_attempt = async move { + reconnect_status.reconnect_attempt = async move { future_instant.await; debug!("Attempting reconnect #{} now.", cur_num); T::establish(ctor_arg).await - }; - - reconnect_status.reconnect_attempt = Box::pin(reconnect_attempt); + } + .boxed(); debug!( "Will perform reconnect attempt #{} in {:?}.", diff --git a/tests/dummy_tests.rs b/tests/dummy_tests.rs index 70e930a..8522151 100644 --- a/tests/dummy_tests.rs +++ b/tests/dummy_tests.rs @@ -1,4 +1,3 @@ -use std::future::Future; use std::io::{self, Error, ErrorKind}; use std::pin::Pin; use std::sync::atomic::{AtomicU8, Ordering}; @@ -33,7 +32,7 @@ impl UnderlyingStream, io::Error> for DummyStreamConnector { type Stream = DummyStream; #[cfg(not(feature = "not-send"))] - fn establish(ctor: DummyCtor) -> Pin> + Send>> { + async fn establish(ctor: DummyCtor) -> io::Result { let mut connect_attempt_outcome_results = ctor.connect_outcomes.lock().unwrap(); let should_succeed = connect_attempt_outcome_results.remove(0); @@ -42,25 +41,9 @@ impl UnderlyingStream, io::Error> for DummyStreamConnector { poll_read_results: ctor.poll_read_results.clone(), }; - Box::pin(async { Ok(dummy_io) }) + Ok(dummy_io) } else { - Box::pin(async { Err(io::Error::new(ErrorKind::NotConnected, "So unfortunate")) }) - } - } - - #[cfg(feature = "not-send")] - fn establish(ctor: DummyCtor) -> Pin>>> { - let mut connect_attempt_outcome_results = ctor.connect_outcomes.lock().unwrap(); - - let should_succeed = connect_attempt_outcome_results.remove(0); - if should_succeed { - let dummy_io = DummyStream { - poll_read_results: ctor.poll_read_results.clone(), - }; - - Box::pin(async { Ok(dummy_io) }) - } else { - Box::pin(async { Err(io::Error::new(ErrorKind::NotConnected, "So unfortunate")) }) + Err(io::Error::new(ErrorKind::NotConnected, "So unfortunate")) } }