Introducing changes to allow for sync to compile

Cargo.toml

@@ -9,7 +9,7 @@ keywords = ["reconnect", "retry", "stubborn", "io", "StubbornTcpStream"]
 repository = "https://github.com/craftytrickster/stubborn-io"
 documentation = "https://docs.rs/stubborn-io"
 readme = "README.md"
- 
+
 [dependencies]
 tokio = { version = "1", features = ["time", "net"] }
 rand = "0.8"

src/tokio/io.rs

@@ -5,6 +5,7 @@ use std::io::{self, ErrorKind, IoSlice};
 use std::marker::PhantomData;
 use std::ops::{Deref, DerefMut};
 use std::pin::Pin;
+use std::sync::{Arc, Mutex};
 use std::task::{Context, Poll};
 use std::time::Duration;
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
@@ -53,12 +54,12 @@ where
 
 struct AttemptsTracker {
  attempt_num: usize,
- retries_remaining: Box<dyn Iterator<Item = Duration> + Send>,
+ retries_remaining: Box<dyn Iterator<Item = Duration> + Send + Sync>,
 }
 
 struct ReconnectStatus<T, C> {
  attempts_tracker: AttemptsTracker,
- reconnect_attempt: Pin<Box<dyn Future<Output = io::Result<T>> + Send>>,
+ reconnect_attempt: Arc<Mutex<Pin<Box<dyn Future<Output = io::Result<T>> + Send>>>>,
  _phantom_data: PhantomData<C>,
 }
 
@@ -73,7 +74,9 @@ where
  attempt_num: 0,
  retries_remaining: (options.retries_to_attempt_fn)(),
  },
- reconnect_attempt: Box::pin(async { unreachable!("Not going to happen") }),
+ reconnect_attempt: Arc::new(Mutex::new(Box::pin(async {
+ unreachable!("Not going to happen")
+ }))),
  _phantom_data: PhantomData,
  }
  }
@@ -244,7 +247,7 @@ where
  T::establish(ctor_arg).await
  };
 
- reconnect_status.reconnect_attempt = Box::pin(reconnect_attempt);
+ reconnect_status.reconnect_attempt = Arc::new(Mutex::new(Box::pin(reconnect_attempt)));
 
  info!(
  "Will perform reconnect attempt #{} in {:?}.",
@@ -259,13 +262,15 @@ where
  let (attempt, attempt_num) = match &mut self.status {
  Status::Connected => unreachable!(),
  Status::Disconnected(ref mut status) => (
- Pin::new(&mut status.reconnect_attempt),
+ status.reconnect_attempt.clone(),
  status.attempts_tracker.attempt_num,
  ),
  Status::FailedAndExhausted => unreachable!(),
  };
 
- match attempt.poll(cx) {
+ let mut attempt = attempt.lock().unwrap();
+
+ match attempt.as_mut().poll(cx) {
  Poll::Ready(Ok(underlying_io)) => {
  info!("Connection re-established");
  cx.waker().wake_by_ref();

tests/dummy_tests.rs

@@ -290,3 +290,25 @@ mod already_connected {
  assert!(msg.unwrap().is_err());
  }
 }
+
+#[tokio::test]
+async fn test_that_works_with_sync() {
+ fn make_framed<T>(_stream: T)
+ where
+ T: AsyncRead + AsyncWrite + Send + Sync + 'static,
+ {
+ let _ = _stream;
+ }
+
+ let options = ReconnectOptions::new();
+ let connect_outcomes = Arc::new(Mutex::new(vec![true]));
+ let ctor = DummyCtor {
+ connect_outcomes,
+ ..DummyCtor::default()
+ };
+ let dummy = StubbornDummy::connect_with_options(ctor, options)
+ .await
+ .unwrap();
+
+ make_framed(dummy);
+}