Trait futures_retry::StreamRetryExt

pub trait StreamRetryExt: TryStream {
    fn retry<F>(self, error_action: F) -> StreamRetry<F, Self>
    where
        Self: Sized,
    { ... }
}

An extention trait for Stream which allows to use StreamRetry in a chain-like manner.

Example

This magic trait allows you to handle errors on streams in a very neat manner:

// ...
use futures_retry::{RetryPolicy, StreamRetryExt};

fn handle_error(e: io::Error) -> RetryPolicy<io::Error> {
  match e.kind() {
    io::ErrorKind::Interrupted => RetryPolicy::Repeat,
    io::ErrorKind::PermissionDenied => RetryPolicy::ForwardError(e),
    _ => RetryPolicy::WaitRetry(Duration::from_millis(5)),
  }
}

async fn serve_connection(stream: TcpStream) {
  // ...
}

#[tokio::main]
async fn main() {
  let mut listener: TcpListener = // ...
  let server = stream::try_unfold(listener, |listener| async move {
    Ok(Some((listener.accept().await?.0, listener)))
  })
  .retry(handle_error)
  .and_then(|(stream, _attempt)| {
    tokio::spawn(serve_connection(stream));
    ok(())
  })
  .try_for_each(|_| ok(()))
  .map_err(|(e, _attempt)| eprintln!("Caught an error {}", e));

  server.await
}

Provided methods

fn retry<F>(self, error_action: F) -> StreamRetry<F, Self> where
    Self: Sized, 

Converts the stream into a retry stream. See StreamRetry::new for details.

Loading content...

Implementors

impl<S: ?Sized> StreamRetryExt for S where
    S: TryStream, 

Loading content...