Trait stream_cancel::StreamExt

pub trait StreamExt: Stream {
    fn take_until<U>(self, until: U) -> TakeUntil<Self, U>
    where
        U: Future<Output = bool>,
        Self: Sized,
    { ... }
}

This Stream extension trait provides a take_until method that terminates the stream once the given future resolves.

Provided methods

fn take_until<U>(self, until: U) -> TakeUntil<Self, U> where
    U: Future<Output = bool>,
    Self: Sized, 

Take elements from this stream until the given future resolves.

This function will take elements from this stream until the given future resolves with true. Once it resolves, the stream will yield None, and produce no further elements.

If the future resolves with false, the stream will be allowed to continue indefinitely.

use stream_cancel::StreamExt;
use tokio::prelude::*;

#[tokio::main]
async fn main() {
    let listener = tokio::net::TcpListener::bind("0.0.0.0:0").await.unwrap();
    let (tx, rx) = tokio::sync::oneshot::channel();

    tokio::spawn(async move {
        let mut incoming = listener.incoming().take_until(rx.map(|_| true));
        while let Some(s) = incoming.next().await.transpose().unwrap() {
            let (mut r, mut w) = s.split();
            tokio::spawn(async move {
                println!("copied {} bytes", r.copy(&mut w).await.unwrap());
            });
        }
    });

    // tell the listener to stop accepting new connections
    tx.send(()).unwrap();
    // the spawned async block will terminate cleanly, allowing main to return
}

Loading content...

Implementors

impl<S> StreamExt for S where
    S: Stream, 

Loading content...