allow AFIT usage

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<String, Result<Message, WsError>, WsError> for MyWs {
 
     // Establishes connection.
     // Additionally, this will be used when reconnect tries are attempted.
-    fn establish(addr: String) -> Pin<Box<dyn Future<Output = Result<Self::Stream, WsError>> + 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<Self::Stream, WsError> {
+        // 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
+MIT

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<Box<dyn Future<Output = Result<Self::Stream, E>>>>;
+    fn establish(ctor_arg: C) -> impl Future<Output = Result<Self::Stream, E>>;
 
     /// 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<Box<dyn Future<Output = Result<Self::Stream, E>> + Send>>;
+    fn establish(ctor_arg: C) -> impl Future<Output = Result<Self::Stream, E>> + 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<Box<dyn Future<Output = Result<T::Stream, E>> + Send>>,
+    reconnect_attempt: BoxFuture<'static, Result<T::Stream, E>>,
     #[cfg(feature = "not-send")]
-    reconnect_attempt: Pin<Box<dyn Future<Output = Result<T::Stream, E>>>>,
+    reconnect_attempt: LocalBoxFuture<'static, Result<T::Stream, E>>,
     _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 {:?}.",

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<DummyCtor, Vec<u8>, io::Error> for DummyStreamConnector {
     type Stream = DummyStream;
 
     #[cfg(not(feature = "not-send"))]
-    fn establish(ctor: DummyCtor) -> Pin<Box<dyn Future<Output = io::Result<DummyStream>> + Send>> {
+    async fn establish(ctor: DummyCtor) -> io::Result<DummyStream> {
         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<DummyCtor, Vec<u8>, 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<Box<dyn Future<Output = io::Result<DummyStream>>>> {
-        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"))
         }
     }