stream-find

A crate that provide trat StreamFind which add method find to any futures::stream::Stream object.

How to use

• use stream_find::StreamFind;
• stream_obj.find(async |item| {// Return Some(item) if found, otherwise return None})

Example

use stream_find::StreamFind;
use futures::stream::{iter, StreamExt};

const START: usize = 0;
const END: usize = 100;
const TARGET: usize = 0;
let mut stream = iter(START..END);
let result = stream.find(async |item| {
    if item == TARGET {
        Some(item)
    } else {
        None
    }
}).await;
assert_eq!(result.unwrap(), TARGET, "Expect to found something.");
assert_eq!(stream.next().await.expect("to yield next value"), TARGET + 1, "Expect stream to be resumable and it immediately stop after it found first match.");

lib.rs:

A crate to allow find method on futures_core::stream::Stream.

This crate provides a trait [StreamFind] which add method StreamFind::find to any struct that implement trait futures_core::stream::Stream and core::marker::Unpin. It required core::marker::Unpin because, as currently is, futures_core::stream::Stream need to be unpin to iterate over it.

The stream object is not consume so the stream can be reuse after the result is found.

The predicate function is different from core::iter::Iterator::find. The predicate function for stream took owned value yield out of stream and pass ownership to predicate method. If predicate function return Option::None, it is the same as return false in core::iter::Iterator::find. If predicate function return Option::Some, it is the same as return true in core::iter::Iterator::find. The value it pass back will be used as a found value so in most case, it should pass back the value it pass in to predicate function.

The reason that it work this way is because, otherwise, it required self referential struct which is required to be pinned. It will required unsafe code to make it work. To avoid using unsafe code completely, The stream yielded value won't be hold by the struct. It passed in as owned value to predicate. If predicate decided that it is not matched, the value is immediately dropped.

Example

use stream_find::StreamFind;
use futures::stream::{iter, StreamExt};
const START: usize = 0;
const END: usize = 100;
const TARGET: usize = 0;
let mut stream = iter(START..END);
let result = stream.find(async |item| {
    if item == TARGET {
        Some(item)
    } else {
        None
    }
}).await;
assert_eq!(result.unwrap(), TARGET, "Expect to found something.");
assert_eq!(stream.next().await.expect("to yield next value"), TARGET + 1, "Expect stream to be resumable and it immediately stop after it found first match.");