Struct SharedStream 

pub struct SharedStream<S>
where
    S: Stream + Unpin,
    S::Item: Clone,
{ /* private fields */ }

A cloneable stream wrapper that allows multiple consumers to share the same stream.

SharedStream wraps any Stream that is Unpin and whose items implement Clone, allowing it to be cloned and consumed by multiple tasks simultaneously. All clones share the same underlying stream state and position.

Examples

Basic usage with concurrent consumers:

use stream_shared::SharedStream;
use futures_util::stream;
use futures_util::StreamExt;

let data = vec!["hello", "world", "from", "rust"];
let stream = stream::iter(data.clone());
let shared_stream = SharedStream::new(stream);

// Create multiple consumers
let consumer1 = shared_stream.clone();
let consumer2 = shared_stream.clone();

// Both will receive all items
let (result1, result2) = tokio::join!(
    consumer1.collect::<Vec<&str>>(),
    consumer2.collect::<Vec<&str>>()
);

assert_eq!(result1, data);
assert_eq!(result2, data);

Cloning after partial consumption:

use stream_shared::SharedStream;
use futures_util::stream;
use futures_util::StreamExt;

let data = vec![1, 2, 3, 4, 5];
let stream = stream::iter(data);
let mut shared_stream = SharedStream::new(stream);

// Consume first item
let first = shared_stream.next().await;
assert_eq!(first, Some(1));

// Clone after partial consumption
let cloned = shared_stream.clone();
let remaining: Vec<i32> = cloned.collect().await;

// Clone only sees remaining items
assert_eq!(remaining, vec![2, 3, 4, 5]);

Requirements

The wrapped stream must satisfy these bounds:

• Stream: The type must implement the Stream trait
• Unpin: Required for safe polling without pinning
• Stream::Item: The item should be clonable

For !Unpin streams, pin them first:

use stream_shared::SharedStream;
use futures_util::stream::{Stream, StreamExt};
use std::pin::Pin;
use std::task::{Context, Poll};

// Create a custom !Unpin stream
#[derive(Clone)]
struct NotUnpinStream {
    data: Vec<i32>,
    index: usize,
    _pin: std::marker::PhantomPinned,
}

impl Stream for NotUnpinStream {
    type Item = i32;
     
    fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        // SAFETY: We only modify fields that are not structurally pinned
        let this = unsafe { self.get_unchecked_mut() };
        if this.index < this.data.len() {
            let item = this.data[this.index];
            this.index += 1;
            Poll::Ready(Some(item))
        } else {
            Poll::Ready(None)
        }
    }
}

let not_unpin_stream = NotUnpinStream {
    data: vec![1, 2, 3],
    index: 0,
    _pin: std::marker::PhantomPinned,
};

// This wouldn't compile: SharedStream::new(not_unpin_stream)
// But this works:
let pinned_stream = Box::pin(not_unpin_stream);
let shared = SharedStream::new(pinned_stream);
let result: Vec<i32> = shared.collect().await;
assert_eq!(result, vec![1, 2, 3]);

Implementations

impl<S> SharedStream<S>

where S: Stream + Unpin, S::Item: Clone,

pub fn new(stream: S) -> Self

Creates a new SharedStream from the given stream.

The stream must implement Unpin and its items must implement Clone. Once created, the SharedStream can be cloned to create multiple consumers that all share the same underlying stream state.

Examples

use stream_shared::SharedStream;
use futures_util::stream;

let data = vec![1, 2, 3, 4, 5];
let stream = stream::iter(data);
let shared_stream = SharedStream::new(stream);

Panics

This method does not panic under normal circumstances.

pub fn stats(&self) -> Stats

Available on crate feature stats only.

Returns the number of active clones of this stream, including the current instance.

This method allows you to track how many shared consumers exist for the stream. The count includes the current instance, so a value of 1 means there are no clones.

Examples

use stream_shared::SharedStream;
use futures_util::stream;

let stream = stream::iter(vec![1, 2, 3]);
let shared = SharedStream::new(stream);
let stats = shared.stats();
assert_eq!(stats.active_clones(), 1); // No clones yet

let clone1 = shared.clone();
assert_eq!(stats.active_clones(), 2); // Original + 1 clone

let clone2 = shared.clone();
assert_eq!(stats.active_clones(), 3); // Original + 2 clones

drop(clone1);
assert_eq!(stats.active_clones(), 2); // Original + 1 clone remaining

Trait Implementations

impl<S> Clone for SharedStream<S>

where S: Stream + Unpin, S::Item: Clone,

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<S> Debug for SharedStream<S>

where S: Stream + Unpin + Debug, S::Item: Clone,

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<S> Drop for SharedStream<S>

where S: Stream + Unpin, S::Item: Clone,

fn drop(&mut self)

Executes the destructor for this type.

impl<S> FusedStream for SharedStream<S>

where S: Stream + Unpin, S::Item: Clone,

fn is_terminated(&self) -> bool

Returns true if the stream should no longer be polled.

impl<S> Stream for SharedStream<S>

where S: Stream + Unpin, S::Item: Clone,

type Item = <S as Stream>::Item

Values yielded by the stream.

fn poll_next( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.