Trait RskitStreamExt 

pub trait RskitStreamExt:
    Stream
    + Sized
    + Send
    + 'static
where
    Self::Item: Send + 'static,
{
    // Provided methods
    fn rmap<O, F, Fut>(
        self,
        f: F,
    ) -> impl Stream<Item = AppResult<O>> + Send + 'static
       where O: Send + 'static,
             F: FnMut(Self::Item) -> Fut + Send + 'static,
             Fut: Future<Output = AppResult<O>> + Send + 'static { ... }
    fn rflatmap<O, F, Fut, St>(
        self,
        f: F,
    ) -> impl Stream<Item = AppResult<O>> + Send + 'static
       where O: Send + 'static,
             F: FnMut(Self::Item) -> Fut + Send + 'static,
             Fut: Future<Output = AppResult<St>> + Send + 'static,
             St: Stream<Item = AppResult<O>> + Send + Unpin + 'static { ... }
    fn rfilter<F>(self, f: F) -> impl Stream<Item = Self::Item> + Send + 'static
       where F: FnMut(&Self::Item) -> bool + Send + 'static { ... }
    fn rdistinct(self) -> impl Stream<Item = Self::Item> + Send + 'static
       where Self::Item: Clone + Eq + Hash { ... }
    fn rtap<F, Fut>(
        self,
        f: F,
    ) -> impl Stream<Item = Self::Item> + Send + 'static
       where F: FnMut(&Self::Item) -> Fut + Send + 'static,
             Fut: Future<Output = ()> + Send + 'static { ... }
    fn rtake(self, n: usize) -> impl Stream<Item = Self::Item> + Send + 'static { ... }
    fn rskip(self, n: usize) -> impl Stream<Item = Self::Item> + Send + 'static { ... }
    fn rpartition<F>(
        self,
        predicate: F,
    ) -> (impl Stream<Item = Self::Item> + Send + 'static, impl Stream<Item = Self::Item> + Send + 'static)
       where F: FnMut(&Self::Item) -> bool + Send + 'static { ... }
    async fn rreduce<Acc, F>(self, init: Acc, f: F) -> Acc
       where Acc: Send + 'static,
             F: FnMut(Acc, Self::Item) -> Acc + Send + 'static { ... }
    fn rparallel<O, F, Fut>(
        self,
        concurrency: usize,
        f: F,
    ) -> impl Stream<Item = AppResult<O>> + Send + 'static
       where O: Send + 'static,
             F: Fn(Self::Item) -> Fut + Clone + Send + Sync + 'static,
             Fut: Future<Output = AppResult<O>> + Send + 'static { ... }
    fn rfan_out<O, F, Fut>(
        self,
        max_branches: usize,
        fns: Vec<F>,
    ) -> impl Stream<Item = AppResult<Vec<O>>> + Send + 'static
       where O: Send + 'static,
             Self::Item: Clone,
             F: Fn(Self::Item) -> Fut + Clone + Send + Sync + 'static,
             Fut: Future<Output = AppResult<O>> + Send + 'static { ... }
    fn rtumbling_window(
        self,
        duration: Duration,
        max_items: usize,
    ) -> impl Stream<Item = Vec<Self::Item>> + Send + 'static { ... }
    fn rsliding_window(
        self,
        size: usize,
        step: usize,
    ) -> impl Stream<Item = Vec<Self::Item>> + Send + 'static
       where Self::Item: Clone { ... }
    fn rbatch(
        self,
        size: usize,
        timeout: Duration,
    ) -> impl Stream<Item = Vec<Self::Item>> + Send + 'static { ... }
    fn rdebounce(
        self,
        delay: Duration,
    ) -> impl Stream<Item = Self::Item> + Send + 'static { ... }
    fn rdebounce_batch(
        self,
        quiet: Duration,
        max_items: usize,
    ) -> impl Stream<Item = Vec<Self::Item>> + Send + 'static { ... }
    fn rthrottle(
        self,
        interval: Duration,
    ) -> impl Stream<Item = Self::Item> + Send + 'static { ... }
    fn rmerge(
        self,
        other: impl Stream<Item = Self::Item> + Send + 'static,
    ) -> impl Stream<Item = Self::Item> + Send + 'static { ... }
}

Extension trait adding rskit-specific operators to any Stream.

Imported with use rskit_stream::RskitStreamExt;.

Provided Methods

fn rmap<O, F, Fut>( self, f: F, ) -> impl Stream<Item = AppResult<O>> + Send + 'static

where O: Send + 'static, F: FnMut(Self::Item) -> Fut + Send + 'static, Fut: Future<Output = AppResult<O>> + Send + 'static,

Async map — apply a fallible async function to each item.

fn rflatmap<O, F, Fut, St>( self, f: F, ) -> impl Stream<Item = AppResult<O>> + Send + 'static

where O: Send + 'static, F: FnMut(Self::Item) -> Fut + Send + 'static, Fut: Future<Output = AppResult<St>> + Send + 'static, St: Stream<Item = AppResult<O>> + Send + Unpin + 'static,

Async flat-map — apply a function that returns a stream and flatten one level.

fn rfilter<F>(self, f: F) -> impl Stream<Item = Self::Item> + Send + 'static

where F: FnMut(&Self::Item) -> bool + Send + 'static,

Keep only items that satisfy the (synchronous) predicate.

fn rdistinct(self) -> impl Stream<Item = Self::Item> + Send + 'static

where Self::Item: Clone + Eq + Hash,

Emit only the first occurrence of each item.

fn rtap<F, Fut>(self, f: F) -> impl Stream<Item = Self::Item> + Send + 'static

where F: FnMut(&Self::Item) -> Fut + Send + 'static, Fut: Future<Output = ()> + Send + 'static,

Side-effect for each item — does not modify the stream.

fn rtake(self, n: usize) -> impl Stream<Item = Self::Item> + Send + 'static

Take the first n items from the stream.

fn rskip(self, n: usize) -> impl Stream<Item = Self::Item> + Send + 'static

Skip the first n items from the stream.

fn rpartition<F>( self, predicate: F, ) -> (impl Stream<Item = Self::Item> + Send + 'static, impl Stream<Item = Self::Item> + Send + 'static)

where F: FnMut(&Self::Item) -> bool + Send + 'static,

Split the stream into (matching, remainder) streams.

async fn rreduce<Acc, F>(self, init: Acc, f: F) -> Acc

where Acc: Send + 'static, F: FnMut(Acc, Self::Item) -> Acc + Send + 'static,

Fold the entire stream into a single value.

fn rparallel<O, F, Fut>( self, concurrency: usize, f: F, ) -> impl Stream<Item = AppResult<O>> + Send + 'static

where O: Send + 'static, F: Fn(Self::Item) -> Fut + Clone + Send + Sync + 'static, Fut: Future<Output = AppResult<O>> + Send + 'static,

Process up to concurrency items in parallel (unordered output).

fn rfan_out<O, F, Fut>( self, max_branches: usize, fns: Vec<F>, ) -> impl Stream<Item = AppResult<Vec<O>>> + Send + 'static

where O: Send + 'static, Self::Item: Clone, F: Fn(Self::Item) -> Fut + Clone + Send + Sync + 'static, Fut: Future<Output = AppResult<O>> + Send + 'static,

Apply multiple functions to the same item and collect all results.

fn rtumbling_window( self, duration: Duration, max_items: usize, ) -> impl Stream<Item = Vec<Self::Item>> + Send + 'static

Collect items into non-overlapping windows bounded by time and item count.

fn rsliding_window( self, size: usize, step: usize, ) -> impl Stream<Item = Vec<Self::Item>> + Send + 'static

where Self::Item: Clone,

Emit sliding windows of size items, advancing by step items each time.

fn rbatch( self, size: usize, timeout: Duration, ) -> impl Stream<Item = Vec<Self::Item>> + Send + 'static

Emit batches of up to size items or when timeout elapses.

fn rdebounce( self, delay: Duration, ) -> impl Stream<Item = Self::Item> + Send + 'static

Emit only when no new item arrives within delay.

fn rdebounce_batch( self, quiet: Duration, max_items: usize, ) -> impl Stream<Item = Vec<Self::Item>> + Send + 'static

Accumulate items into a batch, emitting it once quiet elapses with no new item (trailing-edge debounce that keeps every item, not just the last) or early once max_items accumulate.

fn rthrottle( self, interval: Duration, ) -> impl Stream<Item = Self::Item> + Send + 'static

Emit at most one item per interval.

fn rmerge( self, other: impl Stream<Item = Self::Item> + Send + 'static, ) -> impl Stream<Item = Self::Item> + Send + 'static

Merge this stream with another, yielding items from whichever is ready first.

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl<S> RskitStreamExt for S

where S: Stream + Send + 'static, S::Item: Send + 'static,