Fix io disconnect

.github/workflows/pr-validator.yml

@@ -10,11 +10,8 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-      - uses: actions/checkout@v1
-      - uses: actions-rs/toolchain@v1
-        with:
-          toolchain: stable
-
+      - uses: actions/[email protected]
+      - uses: hecrj/[email protected]
       - name: Format check
         run: cargo fmt  -- --check
       - name: Build and Lint

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "sdre-stubborn-io"
-version = "0.4.4"
+version = "0.5.0"
 authors = ["David Raifaizen <[email protected]>", "Fred Clausen"]
 edition = "2021"
 description = "Forked from https://github.com/craftytrickster/stubborn-io. io traits/structs that automatically recover from potential disconnections/interruptions."
@@ -11,11 +11,11 @@ documentation = "https://docs.rs/sdre-stubborn-io"
 readme = "README.md"
 
 [dependencies]
-tokio = { version = "1.34.0", features = ["time", "net"] }
+tokio = { version = "1.35.1", features = ["time", "net"] }
 log = "0.4.20"
 rand = "0.8.5"
 
 [dev-dependencies]
-tokio = { version = "1.34.0", features = ["macros", "rt", "fs", "io-util"] }
+tokio = { version = "1.35.1", features = ["macros", "rt", "fs", "io-util"] }
 tokio-util = { version = "0.7.10", features = ["codec"] }
-futures = "0.3.29"
+futures = "0.3.30"

README.md

@@ -15,7 +15,7 @@ This project has been forked from [stubborn-io](https://github.com/craftytrickst
 ## Documentation
 
 API Documentation, examples and motivations can be found here -
-(Rust Docs)<https://docs.rs/stubborn-io> .
+(Rust Docs)<https://docs.rs/sdre-stubborn-io> .
 
 Only change to the documentation in this fork will be the addition of the `ReconnectionOptions` struct, which adds `with_connection_name(name: &str)` as a method to the `StubbornTcpStream` struct. This allows for the naming of the connection, which is useful for logging purposes.
 

src/config.rs

@@ -26,6 +26,10 @@ pub struct ReconnectOptions {
     pub on_connect_fail_callback: Box<dyn Fn() + Send + Sync>,
 
     pub connection_name: String,
+
+    /// If this is set to false (default), then the StubbornIo will NOT block
+    /// On write failures.
+    pub block_on_write_failures: bool,
 }
 
 impl ReconnectOptions {
@@ -41,6 +45,7 @@ impl ReconnectOptions {
             on_disconnect_callback: Box::new(|| {}),
             on_connect_fail_callback: Box::new(|| {}),
             connection_name: String::new(),
+            block_on_write_failures: false,
         }
     }
 
@@ -50,7 +55,7 @@ impl ReconnectOptions {
     ///
     /// ```
     /// use std::time::Duration;
-    /// use stubborn_io::ReconnectOptions;
+    /// use sdre_stubborn_io::ReconnectOptions;
     ///
     /// // With the below vector, the stubborn-io item will try to reconnect three times,
     /// // waiting 2 seconds between each attempt. Once all three tries are exhausted,
@@ -97,4 +102,9 @@ impl ReconnectOptions {
         self.connection_name = name.into();
         self
     }
+
+    pub fn with_block_on_write_failures(mut self, value: bool) -> Self {
+        self.block_on_write_failures = value;
+        self
+    }
 }

src/lib.rs

@@ -24,7 +24,7 @@
 //! use std::future::Future;
 //! use std::path::PathBuf;
 //! use std::pin::Pin;
-//! use stubborn_io::tokio::{StubbornIo, UnderlyingIo};
+//! use sdre_stubborn_io::tokio::{StubbornIo, UnderlyingIo};
 //! use tokio::fs::File;
 //!
 //! struct MyFile(File); // Struct must implement AsyncRead + AsyncWrite

src/strategies.rs

@@ -7,7 +7,7 @@ use std::time::Duration;
 ///
 /// ```
 /// use std::time::Duration;
-/// use stubborn_io::{ReconnectOptions, strategies::ExpBackoffStrategy};
+/// use sdre_stubborn_io::{ReconnectOptions, strategies::ExpBackoffStrategy};
 ///
 /// // With the below strategy, the stubborn-io item will try to reconnect infinitely,
 /// // waiting an exponentially increasing (by 2) value with 5% random jitter. Once the

src/tokio/io.rs

@@ -1,5 +1,5 @@
 use crate::config::ReconnectOptions;
-use log::{error, info};
+use log::{error, info, warn};
 use std::future::Future;
 use std::io::{self, ErrorKind, IoSlice};
 use std::marker::PhantomData;
@@ -95,12 +95,24 @@ enum Status<T, C> {
     FailedAndExhausted, // the way one feels after programming in dynamically typed languages
 }
 
+#[inline]
+fn poll_err<T>(
+    kind: ErrorKind,
+    reason: impl Into<Box<dyn std::error::Error + Send + Sync>>,
+) -> Poll<io::Result<T>> {
+    let io_err = io::Error::new(kind, reason);
+    Poll::Ready(Err(io_err))
+}
+
 fn exhausted_err<T>() -> Poll<io::Result<T>> {
-    let io_err = io::Error::new(
+    poll_err(
         ErrorKind::NotConnected,
         "Disconnected. Connection attempts have been exhausted.",
-    );
-    Poll::Ready(Err(io_err))
+    )
+}
+
+fn disconnected_err<T>() -> Poll<io::Result<T>> {
+    poll_err(ErrorKind::NotConnected, "Underlying I/O is disconnected.")
 }
 
 impl<T, C> Deref for StubbornIo<T, C> {
@@ -147,6 +159,10 @@ where
         self.options.connection_name.format_name()
     }
 
+    pub fn get_block_on_write_failures(&self) -> bool {
+        self.options.block_on_write_failures
+    }
+
     pub async fn connect_with_options(ctor_arg: C, options: ReconnectOptions) -> io::Result<Self> {
         let tcp = match T::establish(ctor_arg.clone()).await {
             Ok(tcp) => {
@@ -379,6 +395,10 @@ where
     T: UnderlyingIo<C> + AsyncWrite,
     C: Clone + Send + Unpin + 'static,
 {
+    /// Method for writing to the underlying IO item.
+    /// If the write results in a disconnect. If ReconectOptions::block_on_write_failures is true,
+    /// Poll::Pending is returned to the caller and the buffer is held. Otherwise, the write is skipped
+    /// No error is returned to the caller.
     fn poll_write(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
@@ -389,15 +409,43 @@ where
                 let poll = AsyncWrite::poll_write(Pin::new(&mut self.underlying_io), cx, buf);
 
                 if self.is_write_disconnect_detected(&poll) {
-                    self.on_disconnect(cx);
-                    Poll::Pending
+                    if !self.get_block_on_write_failures() {
+                        error!(
+                            "{}Write disconnect detected. Skipping message",
+                            &self.get_connection_name()
+                        );
+
+                        self.on_disconnect(cx);
+                        Poll::Ready(Ok(buf.len()))
+                    } else {
+                        warn!(
+                            "{}Write disconnect detected. Blocking on write",
+                            &self.get_connection_name()
+                        );
+                        self.on_disconnect(cx);
+                        Poll::Pending
+                    }
                 } else {
                     poll
                 }
             }
             Status::Disconnected(_) => {
-                self.poll_disconnect(cx);
-                Poll::Pending
+                if !self.get_block_on_write_failures() {
+                    error!(
+                        "{}Write disconnect detected. Skipping Message",
+                        &self.get_connection_name()
+                    );
+
+                    self.poll_disconnect(cx);
+                    Poll::Ready(Ok(buf.len()))
+                } else {
+                    warn!(
+                        "{}Write disconnect detected. Blocking on write",
+                        &self.get_connection_name()
+                    );
+                    self.poll_disconnect(cx);
+                    Poll::Pending
+                }
             }
             Status::FailedAndExhausted => exhausted_err(),
         }
@@ -434,11 +482,15 @@ where
 
                 poll
             }
-            Status::Disconnected(_) => Poll::Pending,
+            Status::Disconnected(_) => disconnected_err(),
             Status::FailedAndExhausted => exhausted_err(),
         }
     }
 
+    /// Method for writing to the underlying IO item.
+    /// If the write results in a disconnect. If ReconectOptions::block_on_write_failures is true,
+    /// Poll::Pending is returned to the caller and the buffer is held. Otherwise, the write is skipped
+    /// No error is returned to the caller.
     fn poll_write_vectored(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
@@ -450,15 +502,43 @@ where
                     AsyncWrite::poll_write_vectored(Pin::new(&mut self.underlying_io), cx, bufs);
 
                 if self.is_write_disconnect_detected(&poll) {
-                    self.on_disconnect(cx);
-                    Poll::Pending
+                    if !self.get_block_on_write_failures() {
+                        error!(
+                            "{}Write disconnect detected. Skipping message",
+                            &self.get_connection_name()
+                        );
+
+                        self.on_disconnect(cx);
+                        Poll::Ready(Ok(bufs.iter().map(|buf| buf.len()).sum()))
+                    } else {
+                        warn!(
+                            "{}Write disconnect detected. Blocking on write",
+                            &self.get_connection_name()
+                        );
+                        self.on_disconnect(cx);
+                        Poll::Pending
+                    }
                 } else {
                     poll
                 }
             }
             Status::Disconnected(_) => {
-                self.poll_disconnect(cx);
-                Poll::Pending
+                if !self.get_block_on_write_failures() {
+                    error!(
+                        "{}Write disconnect detected. Skipping Message",
+                        &self.get_connection_name()
+                    );
+
+                    self.poll_disconnect(cx);
+                    Poll::Ready(Ok(bufs.iter().map(|buf| buf.len()).sum()))
+                } else {
+                    warn!(
+                        "{}Write disconnect detected. Blocking on write",
+                        &self.get_connection_name()
+                    );
+                    self.poll_disconnect(cx);
+                    Poll::Pending
+                }
             }
             Status::FailedAndExhausted => exhausted_err(),
         }

src/tokio/tcp.rs

@@ -17,7 +17,7 @@ where
 /// distinction that it will automatically attempt to reconnect in the face of connectivity failures.
 ///
 /// ```
-/// use stubborn_io::StubbornTcpStream;
+/// use sdre_stubborn_io::StubbornTcpStream;
 /// use tokio::io::AsyncWriteExt;
 ///
 /// let addr = "localhost:8080";