Trait StreamExt 

pub trait StreamExt: Stream {
    // Provided methods
    fn next(&mut self) -> Next<'_, Self> 
       where Self: Unpin { ... }
    fn map<T, F>(self, f: F) -> Map<Self, F>
       where Self: Sized,
             F: FnMut(Self::Item) -> T { ... }
    fn then<Fut, F>(self, f: F) -> Then<Self, Fut, F>
       where Self: Sized,
             F: FnMut(Self::Item) -> Fut,
             Fut: Future { ... }
    fn chain<S2>(self, other: S2) -> Chain<Self, S2>
       where Self: Sized,
             S2: Stream<Item = Self::Item> { ... }
    fn merge(self, other: Self) -> Merge<Self>
       where Self: Sized { ... }
    fn zip<S2>(self, other: S2) -> Zip<Self, S2>
       where Self: Sized,
             S2: Stream { ... }
    fn filter<P>(self, predicate: P) -> Filter<Self, P>
       where Self: Sized,
             P: FnMut(&Self::Item) -> bool { ... }
    fn filter_map<T, F>(self, f: F) -> FilterMap<Self, F>
       where Self: Sized,
             F: FnMut(Self::Item) -> Option<T> { ... }
    fn partition<P>(
        self,
        predicate: P,
        lane_capacity: usize,
    ) -> (Partition<Self, P>, Partition<Self, P>)
       where Self: Sized + Unpin,
             P: FnMut(&Self::Item) -> bool { ... }
    fn take(self, n: usize) -> Take<Self>
       where Self: Sized { ... }
    fn take_while<P>(self, predicate: P) -> TakeWhile<Self, P>
       where Self: Sized,
             P: FnMut(&Self::Item) -> bool { ... }
    fn skip(self, n: usize) -> Skip<Self>
       where Self: Sized { ... }
    fn skip_while<P>(self, predicate: P) -> SkipWhile<Self, P>
       where Self: Sized,
             P: FnMut(&Self::Item) -> bool { ... }
    fn enumerate(self) -> Enumerate<Self>
       where Self: Sized { ... }
    fn fuse(self) -> Fuse<Self>
       where Self: Sized { ... }
    fn inspect<F>(self, f: F) -> Inspect<Self, F>
       where Self: Sized,
             F: FnMut(&Self::Item) { ... }
    fn buffered(self, n: usize) -> Buffered<Self>
       where Self: Sized,
             Self::Item: Future { ... }
    fn buffer_unordered(self, n: usize) -> BufferUnordered<Self>
       where Self: Sized,
             Self::Item: Future { ... }
    fn try_buffered(self, n: usize) -> TryBuffered<Self>
       where Self: Sized,
             Self::Item: Future { ... }
    fn collect<C>(self) -> Collect<Self, C> 
       where Self: Sized,
             C: Default + Extend<Self::Item> { ... }
    fn collect_into<C>(self, collection: C) -> Collect<Self, C> 
       where Self: Sized,
             C: Default + Extend<Self::Item> { ... }
    fn chunks(self, size: usize) -> Chunks<Self>
       where Self: Sized { ... }
    fn ready_chunks(self, size: usize) -> ReadyChunks<Self>
       where Self: Sized { ... }
    fn fold<Acc, F>(self, init: Acc, f: F) -> Fold<Self, F, Acc> 
       where Self: Sized,
             F: FnMut(Acc, Self::Item) -> Acc { ... }
    fn for_each<F>(self, f: F) -> ForEach<Self, F> 
       where Self: Sized,
             F: FnMut(Self::Item) { ... }
    fn for_each_async<F, Fut>(self, f: F) -> ForEachAsync<Self, F, Fut> 
       where Self: Sized,
             F: FnMut(Self::Item) -> Fut,
             Fut: Future<Output = ()> { ... }
    fn count(self) -> Count<Self> 
       where Self: Sized { ... }
    fn any<P>(self, predicate: P) -> Any<Self, P> 
       where Self: Sized,
             P: FnMut(&Self::Item) -> bool { ... }
    fn all<P>(self, predicate: P) -> All<Self, P> 
       where Self: Sized,
             P: FnMut(&Self::Item) -> bool { ... }
    fn try_collect<T, E, C>(self) -> TryCollect<Self, C> 
       where Self: Stream<Item = Result<T, E>> + Sized,
             C: Default + Extend<T> { ... }
    fn try_fold<T, E, Acc, F>(self, init: Acc, f: F) -> TryFold<Self, F, Acc> 
       where Self: Stream<Item = Result<T, E>> + Sized,
             F: FnMut(Acc, T) -> Result<Acc, E> { ... }
    fn try_for_each<F, E>(self, f: F) -> TryForEach<Self, F> 
       where Self: Sized,
             F: FnMut(Self::Item) -> Result<(), E> { ... }
    fn scan<St, B, F>(self, initial_state: St, f: F) -> Scan<Self, St, F>
       where Self: Sized,
             F: FnMut(&mut St, Self::Item) -> Option<B> { ... }
    fn peekable(self) -> Peekable<Self>
       where Self: Sized { ... }
    fn throttle(self, period: Duration) -> Throttle<Self>
       where Self: Sized { ... }
    fn debounce(self, period: Duration) -> Debounce<Self>
       where Self: Sized,
             Self::Item: Unpin { ... }
}

Extension trait providing combinator methods for streams.

This trait is automatically implemented for all types that implement Stream. Consuming adapters require Self: Sized; borrowed terminal operations may add Unpin, and concurrent region operations add their own Send and lifetime bounds. Implementing StreamExt does not change the marker traits or lifetime of the underlying stream.

Provided Methods

fn next(&mut self) -> Next<'_, Self>

where Self: Unpin,

Returns the next item from the stream.

Dropping the returned future before it resolves releases the mutable borrow, so the stream can be polled again. It does not roll back state changes made by an earlier poll_next call; losslessness therefore depends on the underlying stream’s documented cancellation contract. Address-sensitive (!Unpin) streams must be pinned and polled through Stream::poll_next instead of this convenience method.

use asupersync::stream::{Stream, StreamExt};
use std::marker::PhantomPinned;
use std::pin::Pin;
use std::task::{Context, Poll};

struct AddressSensitive {
    _pin: PhantomPinned,
}

impl Stream for AddressSensitive {
    type Item = ();

    fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<()>> {
        let _ = self;
        Poll::Ready(None)
    }
}

let mut stream = AddressSensitive { _pin: PhantomPinned };
let _next = stream.next(); // `AddressSensitive` does not implement `Unpin`.

fn map<T, F>(self, f: F) -> Map<Self, F>

where Self: Sized, F: FnMut(Self::Item) -> T,

Transforms each item using a closure.

fn then<Fut, F>(self, f: F) -> Then<Self, Fut, F>

where Self: Sized, F: FnMut(Self::Item) -> Fut, Fut: Future,

Transforms each item using an async closure.

fn chain<S2>(self, other: S2) -> Chain<Self, S2>

where Self: Sized, S2: Stream<Item = Self::Item>,

Chains this stream with another stream.

fn merge(self, other: Self) -> Merge<Self>

where Self: Sized,

Interleaves this stream with another stream of the same concrete type.

For heterogeneous stream types, use chain, zip, or the free merge function with an iterator of streams.

fn zip<S2>(self, other: S2) -> Zip<Self, S2>

where Self: Sized, S2: Stream,

Zips this stream with another stream, yielding pairs.

fn filter<P>(self, predicate: P) -> Filter<Self, P>

where Self: Sized, P: FnMut(&Self::Item) -> bool,

Yields only items that match the predicate.

fn filter_map<T, F>(self, f: F) -> FilterMap<Self, F>

where Self: Sized, F: FnMut(Self::Item) -> Option<T>,

Filters and transforms items in one step.

fn partition<P>( self, predicate: P, lane_capacity: usize, ) -> (Partition<Self, P>, Partition<Self, P>)

where Self: Sized + Unpin, P: FnMut(&Self::Item) -> bool,

Splits this stream into two by predicate.

The first returned stream yields items for which predicate returned true, the second yields the rest. Each item is delivered to exactly one half; the predicate runs once per item.

lane_capacity bounds how many items may be buffered for the half that is not currently being polled. Once that buffer is full, the polling half stalls until its peer drains — so both halves must be consumed, or the unwanted one dropped. See partition for the full backpressure and wakeup contract.

Panics

Panics if lane_capacity is zero.

fn take(self, n: usize) -> Take<Self>

where Self: Sized,

Takes the first n items.

fn take_while<P>(self, predicate: P) -> TakeWhile<Self, P>

where Self: Sized, P: FnMut(&Self::Item) -> bool,

Takes items while the predicate is true.

fn skip(self, n: usize) -> Skip<Self>

where Self: Sized,

Skips the first n items.

fn skip_while<P>(self, predicate: P) -> SkipWhile<Self, P>

where Self: Sized, P: FnMut(&Self::Item) -> bool,

Skips items while the predicate is true.

fn enumerate(self) -> Enumerate<Self>

where Self: Sized,

Enumerates items with their index.

fn fuse(self) -> Fuse<Self>

where Self: Sized,

Fuses the stream to handle None gracefully.

fn inspect<F>(self, f: F) -> Inspect<Self, F>

where Self: Sized, F: FnMut(&Self::Item),

Inspects items without modifying the stream.

fn buffered(self, n: usize) -> Buffered<Self>

where Self: Sized, Self::Item: Future,

Buffers up to n futures, preserving output order.

fn buffer_unordered(self, n: usize) -> BufferUnordered<Self>

where Self: Sized, Self::Item: Future,

Buffers up to n futures, yielding results as they complete.

fn try_buffered(self, n: usize) -> TryBuffered<Self>

where Self: Sized, Self::Item: Future,

Buffers up to n fallible futures in order, stopping at the first Err.

Outputs are yielded in source order, so the terminating error is the first Err in the source sequence rather than the first to complete. That makes the outcome independent of completion timing.

In-flight futures are dropped when the stream short-circuits; they are plain futures, not region tasks. Use try_for_each_concurrent when the per-item work holds obligations and must be drained instead of dropped.

Example

use asupersync::stream::{iter, StreamExt};

async fn first_four(jobs: Vec<Job>) -> Result<Vec<Out>, Error> {
    // 4 jobs run at once; results arrive in job order, and the first
    // failing job in that order ends the stream.
    iter(jobs).map(run).try_buffered(4).try_collect().await
}

fn collect<C>(self) -> Collect<Self, C>

where Self: Sized, C: Default + Extend<Self::Item>,

Collects all items into a collection.

fn collect_into<C>(self, collection: C) -> Collect<Self, C>

where Self: Sized, C: Default + Extend<Self::Item>,

Collects all items into collection, reusing its existing allocation.

This is collect with a caller-supplied starting collection instead of C::default(). Use it to append to an existing buffer, or to recycle one allocation across repeated drains rather than allocating a fresh collection each time.

Items already present in collection are preserved; stream items are appended via Extend.

Example

use asupersync::stream::{iter, StreamExt};

async fn drain_into(buf: Vec<i32>) -> Vec<i32> {
    // Reuses `buf`'s allocation instead of allocating a fresh Vec, and
    // appends after whatever it already held.
    iter(vec![4, 5, 6]).collect_into(buf).await
}

fn chunks(self, size: usize) -> Chunks<Self>

where Self: Sized,

Collects items into fixed-size chunks.

fn ready_chunks(self, size: usize) -> ReadyChunks<Self>

where Self: Sized,

Yields immediately available items up to a maximum chunk size.

fn fold<Acc, F>(self, init: Acc, f: F) -> Fold<Self, F, Acc>

where Self: Sized, F: FnMut(Acc, Self::Item) -> Acc,

Folds all items into a single value.

fn for_each<F>(self, f: F) -> ForEach<Self, F>

where Self: Sized, F: FnMut(Self::Item),

Executes a closure for each item.

fn for_each_async<F, Fut>(self, f: F) -> ForEachAsync<Self, F, Fut>

where Self: Sized, F: FnMut(Self::Item) -> Fut, Fut: Future<Output = ()>,

Executes an async closure for each item.

fn count(self) -> Count<Self>

where Self: Sized,

Counts the number of items in the stream.

fn any<P>(self, predicate: P) -> Any<Self, P>

where Self: Sized, P: FnMut(&Self::Item) -> bool,

Checks if any item matches the predicate.

fn all<P>(self, predicate: P) -> All<Self, P>

where Self: Sized, P: FnMut(&Self::Item) -> bool,

Checks if all items match the predicate.

fn try_collect<T, E, C>(self) -> TryCollect<Self, C>

where Self: Stream<Item = Result<T, E>> + Sized, C: Default + Extend<T>,

Collects items from a stream of Results, short-circuiting on error.

fn try_fold<T, E, Acc, F>(self, init: Acc, f: F) -> TryFold<Self, F, Acc>

where Self: Stream<Item = Result<T, E>> + Sized, F: FnMut(Acc, T) -> Result<Acc, E>,

Folds a stream of Results, short-circuiting on error.

fn try_for_each<F, E>(self, f: F) -> TryForEach<Self, F>

where Self: Sized, F: FnMut(Self::Item) -> Result<(), E>,

Executes a fallible closure for each item, short-circuiting on error.

fn scan<St, B, F>(self, initial_state: St, f: F) -> Scan<Self, St, F>

where Self: Sized, F: FnMut(&mut St, Self::Item) -> Option<B>,

Yields intermediate accumulator values, like Iterator::scan.

For each item, calls f(&mut state, item). If f returns Some(value), the value is yielded. If f returns None, the stream terminates.

fn peekable(self) -> Peekable<Self>

where Self: Sized,

Creates a peekable stream that supports looking at the next item without consuming it.

fn throttle(self, period: Duration) -> Throttle<Self>

where Self: Sized,

Rate-limits the stream to at most one item per period.

The first item passes through immediately. Subsequent items that arrive within the suppression window are dropped.

fn debounce(self, period: Duration) -> Debounce<Self>

where Self: Sized, Self::Item: Unpin,

Debounces the stream, emitting only after a quiet period.

When items arrive, they are buffered. If no new item arrives for period, the most recent item is yielded. When the underlying stream ends, any buffered item is flushed immediately.

Dyn Compatibility

This trait is not dyn compatible.

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

Implementors

impl<S: Stream + ?Sized> StreamExt for S