diff --git a/Cargo.lock b/Cargo.lock index 439e5b1..ecffd2c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -268,6 +268,12 @@ dependencies = [ "crypto-common", ] +[[package]] +name = "dyn-clone" +version = "1.0.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d6ef0072f8a535281e4876be788938b528e9a1d43900b82c2569af7da799125" + [[package]] name = "errno" version = "0.3.9" @@ -726,6 +732,7 @@ name = "stream-reconnect" version = "0.4.0-beta.4" dependencies = [ "async-std", + "dyn-clone", "futures", "log", "rand", diff --git a/Cargo.toml b/Cargo.toml index c959b56..19478ab 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,15 +16,17 @@ exclude = [ ] [features] -default = ["tokio"] +default = ["tokio", "std"] +std = ["log/std", "rand/std", "futures/std"] not-send = [] [dependencies] tokio = { version = "1", features = ["time"], optional = true } async-std = { version = "1", optional = true } -log = "0.4" -rand = "0.8" -futures = "0.3" +log = { version = "0.4", default-features = false } +rand = { version = "0.8", default-features = false, features = ["small_rng"] } +futures = { version = "0.3", default-features = false, features = ["alloc"] } +dyn-clone = "1.0" [dev-dependencies] tokio = { version = "1", features = ["macros", "rt", "sync"] } 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/config.rs b/src/config.rs index 060ed7a..969b10f 100644 --- a/src/config.rs +++ b/src/config.rs @@ -2,8 +2,9 @@ //! specifically related to reconnect behavior. use crate::strategies::ExpBackoffStrategy; -use std::sync::Arc; -use std::time::Duration; +use alloc::{boxed::Box, sync::Arc}; +use core::time::Duration; +use futures::future::BoxFuture; pub type DurationIterator = Box + Send + Sync>; @@ -27,6 +28,11 @@ impl ReconnectOptions { pub(crate) fn on_connect_fail_callback(&self) -> &Arc { &self.0.on_connect_fail_callback } + pub(crate) fn sleep_provider( + &self, + ) -> Arc BoxFuture<'static, ()> + Send + Sync> { + self.0.sleep_provider.clone() + } } #[derive(Clone)] @@ -36,6 +42,7 @@ struct Inner { on_connect_callback: Arc, on_disconnect_callback: Arc, on_connect_fail_callback: Arc, + sleep_provider: Arc BoxFuture<'static, ()> + Send + Sync>, } impl ReconnectOptions { @@ -43,6 +50,7 @@ impl ReconnectOptions { /// By default, the retries iterator waits longer and longer between reconnection attempts, /// until it eventually perpetually tries to reconnect every 30 minutes. #[allow(clippy::new_without_default)] + #[cfg(feature = "std")] pub fn new() -> Self { ReconnectOptions(Box::new(Inner { retries_to_attempt_fn: Arc::new(|| Box::new(ExpBackoffStrategy::default().into_iter())), @@ -50,6 +58,35 @@ impl ReconnectOptions { on_connect_callback: Arc::new(|| {}), on_disconnect_callback: Arc::new(|| {}), on_connect_fail_callback: Arc::new(|| {}), + sleep_provider: Arc::new(|duration| { + Box::pin({ + #[cfg(feature = "tokio")] + { + tokio::time::sleep(duration) + } + #[cfg(feature = "async-std")] + { + async_std::task::sleep(delay) + } + }) + }), + })) + } + + /// By default, the [ReconnectStream](crate::ReconnectStream) will not try to reconnect if the first connect attempt fails. + /// By default, the retries iterator waits longer and longer between reconnection attempts, + /// until it eventually perpetually tries to reconnect every 30 minutes. + #[allow(clippy::new_without_default)] + pub fn with_sleep_provider( + sleep_provider: impl Fn(Duration) -> BoxFuture<'static, ()> + Send + Sync + 'static, + ) -> Self { + ReconnectOptions(Box::new(Inner { + retries_to_attempt_fn: Arc::new(|| Box::new(ExpBackoffStrategy::default().into_iter())), + exit_if_first_connect_fails: true, + on_connect_callback: Arc::new(|| {}), + on_disconnect_callback: Arc::new(|| {}), + on_connect_fail_callback: Arc::new(|| {}), + sleep_provider: Arc::new(sleep_provider), })) } diff --git a/src/lib.rs b/src/lib.rs index 6a06494..92ec34e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -123,6 +123,15 @@ //! # fn main() {} //! ``` +#![cfg_attr(not(any(feature = "std", test)), no_std)] +#![warn( + clippy::alloc_instead_of_core, + clippy::std_instead_of_alloc, + clippy::std_instead_of_core +)] + +extern crate alloc; + #[doc(inline)] pub use crate::config::ReconnectOptions; pub use crate::stream::{ReconnectStream, UnderlyingStream}; diff --git a/src/strategies.rs b/src/strategies.rs index 7fdfd3e..778c6ad 100644 --- a/src/strategies.rs +++ b/src/strategies.rs @@ -1,6 +1,18 @@ //! Provides the strategies used in stubborn io items -use rand::{rngs::StdRng, Rng, SeedableRng}; -use std::time::Duration; +use alloc::boxed::Box; +use core::time::Duration; +use dyn_clone::DynClone; +use rand::{Rng, RngCore, SeedableRng}; + +#[cfg(feature = "std")] +use rand::rngs::StdRng; + +#[cfg(not(feature = "std"))] +use rand::rngs::SmallRng; + +pub trait RngCoreClone: RngCore + Send + Sync + DynClone {} +dyn_clone::clone_trait_object!(RngCoreClone); +impl RngCoreClone for R where R: RngCore + Send + Sync + Clone {} /// Type used for defining the exponential backoff strategy. /// # Examples @@ -24,7 +36,7 @@ pub struct ExpBackoffStrategy { max: Option, factor: f64, jitter: f64, - seed: Option, + rng: Option>, } impl ExpBackoffStrategy { @@ -34,7 +46,7 @@ impl ExpBackoffStrategy { max: None, factor, jitter, - seed: None, + rng: None, } } @@ -46,8 +58,22 @@ impl ExpBackoffStrategy { } /// Set the seed used to generate jitter. Otherwise, will set RNG via entropy. - pub fn with_seed(mut self, seed: u64) -> Self { - self.seed = Some(seed); + pub fn with_seed(self, seed: u64) -> Self { + self.with_rng({ + #[cfg(feature = "std")] + { + StdRng::seed_from_u64(seed) + } + #[cfg(not(feature = "std"))] + { + SmallRng::seed_from_u64(seed) + } + }) + } + + /// Set the RNG used to generate jitter. Otherwise, will set RNG via entropy. + pub fn with_rng(mut self, rng: R) -> Self { + self.rng = Some(Box::new(rng)); self } } @@ -59,7 +85,7 @@ impl Default for ExpBackoffStrategy { max: Some(Duration::from_secs(30 * 60)), factor: 2.0, jitter: 0.05, - seed: None, + rng: None, } } } @@ -70,11 +96,18 @@ impl IntoIterator for ExpBackoffStrategy { fn into_iter(self) -> Self::IntoIter { let init = self.min.as_secs_f64(); - let rng = match self.seed { - Some(seed) => StdRng::seed_from_u64(seed), - None => StdRng::from_entropy(), - }; - + let rng = self.rng.clone().unwrap_or_else(|| { + Box::new({ + #[cfg(feature = "std")] + { + StdRng::from_entropy() + } + #[cfg(not(feature = "std"))] + { + SmallRng::from_entropy() + } + }) + }); ExpBackoffIter { strategy: self, init, @@ -89,7 +122,7 @@ pub struct ExpBackoffIter { strategy: ExpBackoffStrategy, init: f64, pow: u32, - rng: StdRng, + rng: Box, } impl Iterator for ExpBackoffIter { @@ -110,7 +143,7 @@ impl Iterator for ExpBackoffIter { #[cfg(test)] mod test { use super::ExpBackoffStrategy; - use std::time::Duration; + use core::time::Duration; #[test] fn test_exponential_backoff_jitter_values() { diff --git a/src/stream.rs b/src/stream.rs index 2c6e7d2..cfcf89d 100644 --- a/src/stream.rs +++ b/src/stream.rs @@ -1,13 +1,15 @@ -use std::error::Error; -use std::future::Future; -use std::iter::once; -use std::marker::PhantomData; -use std::ops::{Deref, DerefMut}; -use std::pin::Pin; -use std::task::{Context, Poll}; -use std::time::Duration; - -use futures::{ready, Sink, Stream}; +use core::error::Error; +use core::future::Future; +use core::iter::once; +use core::marker::PhantomData; +use core::ops::{Deref, DerefMut}; +use core::pin::Pin; +use core::task::{Context, Poll}; +use core::time::Duration; + +use alloc::boxed::Box; +use futures::future::BoxFuture; +use futures::{ready, FutureExt, Sink, Stream}; use log::{debug, error, info}; use crate::config::ReconnectOptions; @@ -24,12 +26,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 +64,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 +82,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, } } @@ -145,6 +147,7 @@ where { /// Connects or creates a handle to the [UnderlyingStream] item, /// using the default reconnect options. + #[cfg(feature = "std")] pub async fn connect(ctor_arg: C) -> Result { let options = ReconnectOptions::new(); Self::connect_with_options(ctor_arg, options).await @@ -181,12 +184,7 @@ where delay ); - #[cfg(feature = "tokio")] - let sleep_fut = tokio::time::sleep(delay); - #[cfg(feature = "async-std")] - let sleep_fut = async_std::task::sleep(delay); - - sleep_fut.await; + options.sleep_provider()(delay).await; debug!("Attempting reconnect #{} now.", counter + 1); } @@ -209,6 +207,8 @@ where } fn on_disconnect(mut self: Pin<&mut Self>, cx: &mut Context) { + let sleep_provider = self.options.sleep_provider(); + match &mut self.status { // initial disconnect Status::Connected => { @@ -238,21 +238,16 @@ where } }; - #[cfg(feature = "tokio")] - let future_instant = tokio::time::sleep(next_duration); - #[cfg(feature = "async-std")] - let future_instant = async_std::task::sleep(next_duration); + let future_instant = sleep_provider(next_duration); 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")) } }