RS2StreamExt

rs2_stream::rs2_stream_ext

Trait RS2StreamExt 

Source 
pub trait RS2StreamExt:
    Stream
    + Sized
    + Unpin
    + Send
    + 'static {

Show 43 methods    // Provided methods
    fn auto_backpressure_rs2(self) -> RS2Stream<Self::Item>
       where Self::Item: Send + 'static { ... }
    fn auto_backpressure_with_rs2(
        self,
        config: BackpressureConfig,
    ) -> RS2Stream<Self::Item>
       where Self::Item: Send + 'static { ... }
    fn map_rs2<U, F>(self, f: F) -> RS2Stream<U>
       where F: FnMut(Self::Item) -> U + Send + 'static,
             U: Send + 'static { ... }
    fn map_parallel_rs2<O, F>(self, f: F) -> RS2Stream<O>
       where F: Fn(Self::Item) -> O + Send + Sync + Clone + 'static,
             Self::Item: Send + 'static,
             O: Send + 'static { ... }
    fn map_parallel_with_concurrency_rs2<O, F>(
        self,
        concurrency: usize,
        f: F,
    ) -> RS2Stream<O>
       where F: Fn(Self::Item) -> O + Send + Sync + Clone + 'static,
             Self::Item: Send + 'static,
             O: Send + 'static { ... }
    fn filter_rs2<F>(self, f: F) -> RS2Stream<Self::Item>
       where F: FnMut(&Self::Item) -> bool + Send + 'static,
             Self::Item: Send + 'static { ... }
    fn flat_map_rs2<U, St, F>(self, f: F) -> RS2Stream<U>
       where F: FnMut(Self::Item) -> St + Send + 'static,
             St: Stream<Item = U> + Send + 'static,
             U: Send + 'static { ... }
    fn eval_map_rs2<U, Fut, F>(self, f: F) -> RS2Stream<U>
       where F: FnMut(Self::Item) -> Fut + Send + 'static,
             Fut: Future<Output = U> + Send + 'static,
             U: Send + 'static { ... }
    fn merge_rs2(self, other: RS2Stream<Self::Item>) -> RS2Stream<Self::Item>
       where Self::Item: Send + 'static { ... }
    fn zip_rs2<U>(self, other: RS2Stream<U>) -> RS2Stream<(Self::Item, U)>
       where Self::Item: Send + 'static,
             U: Send + 'static { ... }
    fn zip_with_rs2<U, O, F>(self, other: RS2Stream<U>, f: F) -> RS2Stream<O>
       where Self::Item: Send + 'static,
             U: Send + 'static,
             O: Send + 'static,
             F: FnMut(Self::Item, U) -> O + Send + 'static { ... }
    fn throttle_rs2(self, duration: Duration) -> RS2Stream<Self::Item>
       where Self::Item: Send + 'static { ... }
    fn debounce_rs2(self, duration: Duration) -> RS2Stream<Self::Item>
       where Self::Item: Send + 'static { ... }
    fn sample_rs2(self, interval: Duration) -> RS2Stream<Self::Item>
       where Self::Item: Clone + Send + 'static { ... }
    fn par_eval_map_rs2<U, Fut, F>(
        self,
        concurrency: usize,
        f: F,
    ) -> RS2Stream<U>
       where F: FnMut(Self::Item) -> Fut + Send + 'static,
             Fut: Future<Output = U> + Send + 'static,
             U: Send + 'static,
             Self::Item: Send + 'static { ... }
    fn par_eval_map_unordered_rs2<U, Fut, F>(
        self,
        concurrency: usize,
        f: F,
    ) -> RS2Stream<U>
       where F: FnMut(Self::Item) -> Fut + Send + 'static,
             Fut: Future<Output = U> + Send + 'static,
             U: Send + 'static,
             Self::Item: Send + 'static { ... }
    fn par_join_rs2<S, O>(self, concurrency: usize) -> RS2Stream<O>
       where Self: Stream<Item = S>,
             S: Stream<Item = O> + Send + 'static + Unpin,
             O: Send + 'static { ... }
    fn timeout_rs2(
        self,
        duration: Duration,
    ) -> RS2Stream<StreamResult<Self::Item>>
       where Self::Item: Send + 'static { ... }
    fn prefetch_rs2(self, prefetch_count: usize) -> RS2Stream<Self::Item>
       where Self::Item: Send + 'static { ... }
    fn distinct_until_changed_rs2(self) -> RS2Stream<Self::Item>
       where Self::Item: Clone + Send + PartialEq + 'static { ... }
    fn distinct_until_changed_by_rs2<F>(self, eq: F) -> RS2Stream<Self::Item>
       where Self::Item: Clone + Send + 'static,
             F: FnMut(&Self::Item, &Self::Item) -> bool + Send + 'static { ... }
    fn interrupt_when_rs2<F>(self, signal: F) -> RS2Stream<Self::Item>
       where Self::Item: Send + 'static,
             F: Future<Output = ()> + Send + 'static { ... }
    fn take_while_rs2<F, Fut>(self, predicate: F) -> RS2Stream<Self::Item>
       where F: FnMut(&Self::Item) -> Fut + Send + 'static,
             Fut: Future<Output = bool> + Send + 'static,
             Self::Item: Send + 'static { ... }
    fn drop_while_rs2<F, Fut>(self, predicate: F) -> RS2Stream<Self::Item>
       where F: FnMut(&Self::Item) -> Fut + Send + 'static,
             Fut: Future<Output = bool> + Send + 'static,
             Self::Item: Send + 'static { ... }
    fn group_adjacent_by_rs2<K, F>(
        self,
        key_fn: F,
    ) -> RS2Stream<(K, Vec<Self::Item>)>
       where Self::Item: Clone + Send + 'static,
             K: Eq + Clone + Send + 'static,
             F: FnMut(&Self::Item) -> K + Send + 'static { ... }
    fn group_by_rs2<K, F>(self, key_fn: F) -> RS2Stream<(K, Vec<Self::Item>)>
       where Self::Item: Clone + Send + 'static,
             K: Eq + Clone + Send + 'static,
             F: FnMut(&Self::Item) -> K + Send + 'static { ... }
    fn fold_rs2<A, F, Fut>(self, init: A, f: F) -> impl Future<Output = A>
       where F: FnMut(A, Self::Item) -> Fut + Send + 'static,
             Fut: Future<Output = A> + Send + 'static,
             Self::Item: Send + 'static,
             A: Send + 'static { ... }
    fn scan_rs2<U, F>(self, init: U, f: F) -> RS2Stream<U>
       where F: FnMut(U, Self::Item) -> U + Send + 'static,
             Self::Item: Send + 'static,
             U: Clone + Send + 'static { ... }
    fn for_each_rs2<F, Fut>(self, f: F) -> impl Future<Output = ()>
       where F: FnMut(Self::Item) -> Fut + Send + 'static,
             Fut: Future<Output = ()> + Send + 'static,
             Self::Item: Send + 'static { ... }
    fn take_rs2(self, n: usize) -> RS2Stream<Self::Item>
       where Self::Item: Send + 'static { ... }
    fn drop_rs2(self, n: usize) -> RS2Stream<Self::Item>
       where Self::Item: Send + 'static { ... }
    fn skip_rs2(self, n: usize) -> RS2Stream<Self::Item>
       where Self::Item: Send + 'static { ... }
    fn either_rs2(self, other: RS2Stream<Self::Item>) -> RS2Stream<Self::Item>
       where Self::Item: Send + 'static { ... }
    fn collect_rs2<B>(self) -> impl Future<Output = B>
       where B: Default + Extend<Self::Item> + Send + 'static,
             Self::Item: Send + 'static { ... }
    fn collect_with_config_rs2<B>(
        self,
        config: BufferConfig,
    ) -> impl Future<Output = B>
       where B: Default + Extend<Self::Item> + Send + 'static,
             Self::Item: Send + 'static { ... }
    fn sliding_window_rs2(self, size: usize) -> RS2Stream<Vec<Self::Item>>
       where Self::Item: Clone + Send + 'static { ... }
    fn batch_process_rs2<U, F>(
        self,
        batch_size: usize,
        processor: F,
    ) -> RS2Stream<U>
       where F: FnMut(Vec<Self::Item>) -> Vec<U> + Send + 'static,
             Self::Item: Send + 'static,
             U: Send + 'static { ... }
    fn with_metrics_rs2(
        self,
        name: String,
        health_thresholds: HealthThresholds,
    ) -> (RS2Stream<Self::Item>, Arc<Mutex<StreamMetrics>>)
       where Self::Item: Send + 'static { ... }
    fn interleave_rs2<S>(self, streams: Vec<S>) -> RS2Stream<Self::Item>
       where S: Stream<Item = Self::Item> + Send + 'static + Unpin,
             Self::Item: Send + 'static { ... }
    fn chunk_rs2(self, size: usize) -> RS2Stream<Vec<Self::Item>>
       where Self::Item: Send + 'static { ... }
    fn tick_rs<O>(self, period: Duration, item: O) -> RS2Stream<O>
       where O: Clone + Send + 'static { ... }
    fn bracket_rs<A, O, St, FAcq, FUse, FRel, R>(
        self,
        acquire: FAcq,
        use_fn: FUse,
        release: FRel,
    ) -> RS2Stream<O>
       where FAcq: Future<Output = A> + Send + 'static,
             FUse: FnOnce(A) -> St + Send + 'static,
             St: Stream<Item = O> + Send + 'static,
             FRel: FnOnce(A) -> R + Send + 'static,
             R: Future<Output = ()> + Send + 'static,
             O: Send + 'static,
             A: Clone + Send + 'static { ... }
    fn with_schema_validation_rs2<V, T>(
        self,
        validator: V,
    ) -> Pin<Box<dyn Stream<Item = T> + Send>>
       where V: SchemaValidator + 'static,
             T: DeserializeOwned + Serialize + Send + 'static,
             Self: Stream<Item = T> + Send + 'static { ... }

}
Expand description

Extension trait providing RS2-like combinators on Streams

Provided Methods§

Source

fn auto_backpressure_rs2(self) -> RS2Stream<Self::Item>
where Self::Item: Send + 'static,

Apply automatic backpressure with default configuration

Source

fn auto_backpressure_with_rs2( self, config: BackpressureConfig, ) -> RS2Stream<Self::Item>
where Self::Item: Send + 'static,

Apply automatic backpressure with custom configuration

Source

fn map_rs2<U, F>(self, f: F) -> RS2Stream<U>
where F: FnMut(Self::Item) -> U + Send + 'static, U: Send + 'static,

Map elements of the rs2_stream with a function

Source

fn map_parallel_rs2<O, F>(self, f: F) -> RS2Stream<O>
where F: Fn(Self::Item) -> O + Send + Sync + Clone + 'static, Self::Item: Send + 'static, O: Send + 'static,

Transforms each element of the stream in parallel using all available CPU cores.

This method applies the given synchronous function to each element concurrently, automatically detecting the number of CPU cores and using that as the concurrency limit. Perfect for CPU-bound operations that benefit from parallelization.

§Arguments
  • f - A synchronous function that transforms each stream element. Must be Send + Sync + Clone.
§Returns

A new RS2Stream containing the transformed elements. Order may not be preserved.

§Performance
  • Concurrency: Automatically uses num_cpus::get() concurrent tasks
  • Best for: CPU-intensive computations (math, parsing, compression)
  • Memory: Uses one task per CPU core, moderate memory overhead
  • Backpressure: Inherits from underlying par_eval_map_rs2
§When to Use
  • ✅ CPU-bound work: Mathematical calculations, data parsing, compression
  • ✅ Simple parallelization: Don’t want to think about optimal concurrency
  • ✅ Balanced workloads: Each task takes roughly the same time
  • ❌ I/O-bound work: Use par_eval_map_rs2 with higher concurrency instead
  • ❌ Memory-intensive: May overwhelm system with too many concurrent tasks
§See Also
  • [map_parallel_with_concurrency_rs2] - For custom concurrency control
  • [par_eval_map_rs2] - For async functions and fine-tuned concurrency
Source

fn map_parallel_with_concurrency_rs2<O, F>( self, concurrency: usize, f: F, ) -> RS2Stream<O>
where F: Fn(Self::Item) -> O + Send + Sync + Clone + 'static, Self::Item: Send + 'static, O: Send + 'static,

Transforms each element of the stream in parallel with custom concurrency control.

This method applies the given synchronous function to each element concurrently, using exactly the specified number of concurrent tasks. Ideal when you need precise control over resource usage or when the optimal concurrency differs from CPU count.

§Arguments
  • concurrency - Maximum number of concurrent tasks (must be > 0)
  • f - A synchronous function that transforms each stream element. Must be Send + Sync + Clone.
§Returns

A new RS2Stream containing the transformed elements. Order may not be preserved.

§Performance
  • Concurrency: Uses exactly concurrency concurrent tasks
  • Best for: I/O-bound operations, memory-constrained environments, fine-tuning
  • Memory: Scales with concurrency parameter
  • Backpressure: Inherits from underlying par_eval_map_rs2
§Concurrency Guidelines
Workload TypeRecommended ConcurrencyReasoning
CPU-boundnum_cpus::get()Match CPU cores
I/O-bound50-200Network can handle many concurrent requests
Memory-heavy1-4Prevent out-of-memory errors
Database queries10-50Respect connection pool limits
File I/O4-16Balance throughput vs file handle limits
§When to Use
  • ✅ I/O-bound operations: Network requests, file operations, database queries
  • ✅ Resource constraints: Limited memory, connection pools, rate limits
  • ✅ Performance tuning: Benchmarked optimal concurrency for your workload
  • ✅ Mixed workloads: Some tasks much slower/faster than others
  • ❌ Simple CPU-bound work: Use map_parallel_rs2 for automatic optimization
§Panics

This function will panic if concurrency is 0. Always use a positive value.

§See Also
  • [map_parallel_rs2] - For automatic concurrency based on CPU cores
  • [par_eval_map_rs2] - For async functions with the same concurrency control
Source

fn filter_rs2<F>(self, f: F) -> RS2Stream<Self::Item>
where F: FnMut(&Self::Item) -> bool + Send + 'static, Self::Item: Send + 'static,

Filter elements of the rs2_stream with a predicate

Source

fn flat_map_rs2<U, St, F>(self, f: F) -> RS2Stream<U>
where F: FnMut(Self::Item) -> St + Send + 'static, St: Stream<Item = U> + Send + 'static, U: Send + 'static,

Flat map elements of the rs2_stream with a function that returns a rs2_stream

Source

fn eval_map_rs2<U, Fut, F>(self, f: F) -> RS2Stream<U>
where F: FnMut(Self::Item) -> Fut + Send + 'static, Fut: Future<Output = U> + Send + 'static, U: Send + 'static,

Map elements of the rs2_stream with an async function

Source

fn merge_rs2(self, other: RS2Stream<Self::Item>) -> RS2Stream<Self::Item>
where Self::Item: Send + 'static,

Merge this rs2_stream with another rs2_stream

Source

fn zip_rs2<U>(self, other: RS2Stream<U>) -> RS2Stream<(Self::Item, U)>
where Self::Item: Send + 'static, U: Send + 'static,

Zip this rs2_stream with another rs2_stream

Source

fn zip_with_rs2<U, O, F>(self, other: RS2Stream<U>, f: F) -> RS2Stream<O>
where Self::Item: Send + 'static, U: Send + 'static, O: Send + 'static, F: FnMut(Self::Item, U) -> O + Send + 'static,

Zip this rs2_stream with another rs2_stream, applying a function to each pair

Source

fn throttle_rs2(self, duration: Duration) -> RS2Stream<Self::Item>
where Self::Item: Send + 'static,

Throttle this rs2_stream to emit at most one element per duration

Source

fn debounce_rs2(self, duration: Duration) -> RS2Stream<Self::Item>
where Self::Item: Send + 'static,

Debounce this rs2_stream, only emitting an element after a specified quiet period has passed without receiving another element

Source

fn sample_rs2(self, interval: Duration) -> RS2Stream<Self::Item>
where Self::Item: Clone + Send + 'static,

Sample this rs2_stream at regular intervals, emitting the most recent value

This combinator samples the most recent value from a rs2_stream at a regular interval. It only emits a value if at least one new value has arrived since the last emission. If no new value has arrived during an interval, that interval is skipped.

Source

fn par_eval_map_rs2<U, Fut, F>(self, concurrency: usize, f: F) -> RS2Stream<U>
where F: FnMut(Self::Item) -> Fut + Send + 'static, Fut: Future<Output = U> + Send + 'static, U: Send + 'static, Self::Item: Send + 'static,

Process elements in parallel with bounded concurrency, preserving order

§Use when: par_eval_map_rs2
  • Already have async functions**
  • Need custom concurrency control**
  • Want maximum control/performance**
Source

fn par_eval_map_unordered_rs2<U, Fut, F>( self, concurrency: usize, f: F, ) -> RS2Stream<U>
where F: FnMut(Self::Item) -> Fut + Send + 'static, Fut: Future<Output = U> + Send + 'static, U: Send + 'static, Self::Item: Send + 'static,

Process elements in parallel with bounded concurrency, without preserving order

Source

fn par_join_rs2<S, O>(self, concurrency: usize) -> RS2Stream<O>
where Self: Stream<Item = S>, S: Stream<Item = O> + Send + 'static + Unpin, O: Send + 'static,

Run multiple streams concurrently and combine their outputs

This combinator takes a rs2_stream of streams and a concurrency limit, and runs up to n inner streams concurrently. It emits all elements from the inner streams, and starts new inner streams as others complete.

Source

fn timeout_rs2(self, duration: Duration) -> RS2Stream<StreamResult<Self::Item>>
where Self::Item: Send + 'static,

Add timeout to rs2_stream operations

Source

fn prefetch_rs2(self, prefetch_count: usize) -> RS2Stream<Self::Item>
where Self::Item: Send + 'static,

Prefetch a specified number of elements ahead of consumption

This can improve performance by starting to process the next elements before they’re actually needed.

Source

fn distinct_until_changed_rs2(self) -> RS2Stream<Self::Item>
where Self::Item: Clone + Send + PartialEq + 'static,

Filter out consecutive duplicate elements from this rs2_stream

This combinator only emits elements that are different from the previous element. It uses the default equality operator (==) to compare elements. The first element is always emitted.

Source

fn distinct_until_changed_by_rs2<F>(self, eq: F) -> RS2Stream<Self::Item>
where Self::Item: Clone + Send + 'static, F: FnMut(&Self::Item, &Self::Item) -> bool + Send + 'static,

Filter out consecutive duplicate elements from this rs2_stream using a custom equality function

This combinator only emits elements that are different from the previous element. It uses the provided equality function to compare elements. The first element is always emitted.

Source

fn interrupt_when_rs2<F>(self, signal: F) -> RS2Stream<Self::Item>
where Self::Item: Send + 'static, F: Future<Output = ()> + Send + 'static,

Interrupt this rs2_stream when a signal is received

This combinator stops processing the rs2_stream when the signal future completes. Resources are properly cleaned up when the rs2_stream is interrupted.

Source

fn take_while_rs2<F, Fut>(self, predicate: F) -> RS2Stream<Self::Item>
where F: FnMut(&Self::Item) -> Fut + Send + 'static, Fut: Future<Output = bool> + Send + 'static, Self::Item: Send + 'static,

Take elements from this rs2_stream while a predicate returns true

This combinator yields elements from the stream as long as the predicate returns true. It stops (and does not yield) the first element where the predicate returns false.

Source

fn drop_while_rs2<F, Fut>(self, predicate: F) -> RS2Stream<Self::Item>
where F: FnMut(&Self::Item) -> Fut + Send + 'static, Fut: Future<Output = bool> + Send + 'static, Self::Item: Send + 'static,

Skip elements from this rs2_stream while a predicate returns true

This combinator skips elements from the stream as long as the predicate returns true. Once the predicate returns false, it yields that element and all remaining elements.

Source

fn group_adjacent_by_rs2<K, F>( self, key_fn: F, ) -> RS2Stream<(K, Vec<Self::Item>)>
where Self::Item: Clone + Send + 'static, K: Eq + Clone + Send + 'static, F: FnMut(&Self::Item) -> K + Send + 'static,

Group adjacent elements that share a common key

This combinator groups consecutive elements that produce the same key. It emits groups as they complete (when the key changes or the rs2_stream ends). Each emitted item is a tuple containing the key and a vector of elements.

Source

fn group_by_rs2<K, F>(self, key_fn: F) -> RS2Stream<(K, Vec<Self::Item>)>
where Self::Item: Clone + Send + 'static, K: Eq + Clone + Send + 'static, F: FnMut(&Self::Item) -> K + Send + 'static,

Group consecutive elements that share a common key

This combinator groups consecutive elements that produce the same key. It emits groups as they complete (when the key changes or the rs2_stream ends). Each emitted item is a tuple containing the key and a vector of elements.

Source

fn fold_rs2<A, F, Fut>(self, init: A, f: F) -> impl Future<Output = A>
where F: FnMut(A, Self::Item) -> Fut + Send + 'static, Fut: Future<Output = A> + Send + 'static, Self::Item: Send + 'static, A: Send + 'static,

Fold operation that accumulates a value over a stream

This combinator applies a function to each element in the stream, accumulating a single result. It returns a Future that resolves to the final accumulated value.

Source

fn scan_rs2<U, F>(self, init: U, f: F) -> RS2Stream<U>
where F: FnMut(U, Self::Item) -> U + Send + 'static, Self::Item: Send + 'static, U: Clone + Send + 'static,

Scan operation that applies a function to each element and emits intermediate accumulated values

This combinator is similar to fold but emits each intermediate accumulated value. It applies a function to each element in the stream, accumulating a result and yielding each intermediate accumulated value.

Source

fn for_each_rs2<F, Fut>(self, f: F) -> impl Future<Output = ()>
where F: FnMut(Self::Item) -> Fut + Send + 'static, Fut: Future<Output = ()> + Send + 'static, Self::Item: Send + 'static,

Apply a function to each element in the stream

This combinator applies a function to each element in the stream without accumulating a result. It returns a Future that completes when the stream is exhausted.

Source

fn take_rs2(self, n: usize) -> RS2Stream<Self::Item>
where Self::Item: Send + 'static,

Take the first n elements from the stream

This combinator yields the first n elements from the stream and then stops.

Source

fn drop_rs2(self, n: usize) -> RS2Stream<Self::Item>
where Self::Item: Send + 'static,

Drop the first n elements from the stream

This combinator skips the first n elements from the stream and yields all remaining elements.

Source

fn skip_rs2(self, n: usize) -> RS2Stream<Self::Item>
where Self::Item: Send + 'static,

Skip the first n elements from the stream

This combinator skips the first n elements from the stream and yields all remaining elements.

Source

fn either_rs2(self, other: RS2Stream<Self::Item>) -> RS2Stream<Self::Item>
where Self::Item: Send + 'static,

Select between this rs2_stream and another rs2_stream based on which one produces a value first

This combinator emits values from whichever rs2_stream produces a value first. Once a value is received from one rs2_stream, the other rs2_stream is cancelled. If either rs2_stream completes (returns None), the combinator switches to the other rs2_stream exclusively.

Source

fn collect_rs2<B>(self) -> impl Future<Output = B>
where B: Default + Extend<Self::Item> + Send + 'static, Self::Item: Send + 'static,

Collect all items from the stream into a collection

This combinator collects all items from the stream into a collection of type B. It returns a Future that resolves to the collection.

§Examples
use rs2_stream::rs2::*;
use futures_util::stream::StreamExt;

let stream = from_iter(vec![1, 2, 3, 4, 5]);
let result = stream.collect_rs2::<Vec<_>>().await;
assert_eq!(result, vec![1, 2, 3, 4, 5]);
Source

fn collect_with_config_rs2<B>( self, config: BufferConfig, ) -> impl Future<Output = B>
where B: Default + Extend<Self::Item> + Send + 'static, Self::Item: Send + 'static,

Collect all items from the stream into a collection with custom buffer configuration

This combinator collects all items from the stream into a collection of type B. It returns a Future that resolves to the collection. The buffer configuration allows for optimized memory allocation and growth strategies.

Source

fn sliding_window_rs2(self, size: usize) -> RS2Stream<Vec<Self::Item>>
where Self::Item: Clone + Send + 'static,

Create a sliding window of elements from the stream

This combinator creates a sliding window of the specified size over the stream. It yields a vector of items for each window position.

Source

fn batch_process_rs2<U, F>( self, batch_size: usize, processor: F, ) -> RS2Stream<U>
where F: FnMut(Vec<Self::Item>) -> Vec<U> + Send + 'static, Self::Item: Send + 'static, U: Send + 'static,

Process items in batches for better throughput

This combinator processes items in batches of the specified size, applying the processor function to each batch.

Source

fn with_metrics_rs2( self, name: String, health_thresholds: HealthThresholds, ) -> (RS2Stream<Self::Item>, Arc<Mutex<StreamMetrics>>)
where Self::Item: Send + 'static,

Collect metrics while processing the stream

This combinator collects metrics while processing the stream, returning both the stream and the metrics.

Source

fn interleave_rs2<S>(self, streams: Vec<S>) -> RS2Stream<Self::Item>
where S: Stream<Item = Self::Item> + Send + 'static + Unpin, Self::Item: Send + 'static,

Interleave multiple streams in a round-robin fashion

This combinator takes a vector of streams and interleaves their elements in a round-robin fashion.

Source

fn chunk_rs2(self, size: usize) -> RS2Stream<Vec<Self::Item>>
where Self::Item: Send + 'static,

Chunk the stream into vectors of the specified size

This combinator collects elements from the stream into vectors of the specified size. If the stream ends before a chunk is filled, the final chunk may contain fewer elements.

Source

fn tick_rs<O>(self, period: Duration, item: O) -> RS2Stream<O>
where O: Clone + Send + 'static,

Create a stream that emits values at a fixed rate

This combinator creates a stream that emits the provided item at a fixed rate.

Source

fn bracket_rs<A, O, St, FAcq, FUse, FRel, R>( self, acquire: FAcq, use_fn: FUse, release: FRel, ) -> RS2Stream<O>
where FAcq: Future<Output = A> + Send + 'static, FUse: FnOnce(A) -> St + Send + 'static, St: Stream<Item = O> + Send + 'static, FRel: FnOnce(A) -> R + Send + 'static, R: Future<Output = ()> + Send + 'static, O: Send + 'static, A: Clone + Send + 'static,

Bracket for resource management

This combinator ensures that a resource is properly released after use. It takes three parameters:

  1. A future that acquires a resource
  2. A function that uses the resource and returns a stream
  3. A function that releases the resource
Source

fn with_schema_validation_rs2<V, T>( self, validator: V, ) -> Pin<Box<dyn Stream<Item = T> + Send>>
where V: SchemaValidator + 'static, T: DeserializeOwned + Serialize + Send + 'static, Self: Stream<Item = T> + Send + 'static,

Dyn Compatibility§

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety", so this trait is not object safe.

Implementors§

Source§

impl<S> RS2StreamExt for S
where S: Stream + Sized + Unpin + Send + 'static,