Struct StreamMap 

pub struct StreamMap<K, V> { /* private fields */ }

Merge many streams, tagged by key.

Yields (key, item) in arrival order. Insert and remove at any time. Polling starts at a random entry to reduce fixed-order bias. This does not guarantee a maximum wait time for any stream.

Unlike tokio_stream::StreamMap, keyed ops are O(1) via a hash index. Polling still scans entries, same as upstream.

Example

use keyed_stream_map::StreamMap;
use tokio_stream::{self as stream, StreamExt};

let mut map = StreamMap::new();
map.insert("a", stream::iter(vec![1]));
map.insert("b", stream::iter(vec![2]));
assert_eq!(map.len(), 2);
assert_eq!(map.get(&"a").is_some(), true);
while let Some((k, v)) = map.next().await {
    println!("{k}: {v}");
}

Implementations

impl<K, V> StreamMap<K, V>

pub fn new() -> Self

Empty map. Allocates on first insert.

use keyed_stream_map::StreamMap;
use tokio_stream::Pending;
let map: StreamMap<&str, Pending<()>> = StreamMap::new();
assert!(map.is_empty());

pub fn with_capacity(capacity: usize) -> Self

Empty map that can hold capacity entries without reallocating.

pub fn iter(&self) -> impl Iterator<Item = &(K, V)>

Iterate over (key, stream) pairs.

pub fn iter_mut(&mut self) -> impl Iterator<Item = (&K, &mut V)>

Iterate over keys and mutable references to their streams.

Keys are immutable because the map keeps a separate index of them.

pub fn keys(&self) -> impl Iterator<Item = &K>

All keys.

pub fn values(&self) -> impl Iterator<Item = &V>

All streams.

pub fn values_mut(&mut self) -> impl Iterator<Item = &mut V>

All streams, mutably.

pub fn capacity(&self) -> usize

Room for more entries without reallocating. Lower bound.

pub fn len(&self) -> usize

Count of streams in the map.

pub fn is_empty(&self) -> bool

True when no streams are stored.

pub fn clear(&mut self)

Drop all streams. Keeps allocated memory.

pub fn insert(&mut self, k: K, stream: V) -> Option<V>

where K: Hash + Eq + Clone,

Insert a stream. Returns the old stream when the key existed.

Replace keeps the entry position, so polling order is stable.

pub fn remove<Q>(&mut self, k: &Q) -> Option<V>

where K: Borrow<Q> + Hash + Eq + Clone, Q: Hash + Eq + ?Sized,

Remove by key. Returns the stream when present.

pub fn contains_key<Q>(&self, k: &Q) -> bool

where K: Borrow<Q> + Hash + Eq, Q: Hash + Eq + ?Sized,

True when the key has a stream.

pub fn get<Q>(&self, k: &Q) -> Option<&V>

where K: Borrow<Q> + Hash + Eq, Q: Hash + Eq + ?Sized,

Reference to the stream for k, if present.

pub fn get_mut<Q>(&mut self, k: &Q) -> Option<&mut V>

where K: Borrow<Q> + Hash + Eq, Q: Hash + Eq + ?Sized,

Mutable reference to the stream for k, if present.

impl<K, V> StreamMap<K, V>

where K: Clone + Unpin, V: Stream + Unpin,

pub async fn next_many( &mut self, buffer: &mut Vec<(K, V::Item)>, limit: usize, ) -> usize

where K: Hash + Eq,

Pull up to limit items into buffer. Returns count appended.

Returns 0 only when the map is empty or limit is 0. Cancel safe: dropping the future loses nothing.

pub fn poll_next_many( &mut self, cx: &mut Context<'_>, buffer: &mut Vec<(K, V::Item)>, limit: usize, ) -> Poll<usize>

where K: Hash + Eq,

Poll version of StreamMap::next_many.

Trait Implementations

impl<K: Clone, V: Clone> Clone for StreamMap<K, V>

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<K: Debug, V: Debug> Debug for StreamMap<K, V>

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

Formats the value using the given formatter.

impl<K, V> Default for StreamMap<K, V>

fn default() -> Self

Returns the “default value” for a type.

impl<K, V> Extend<(K, V)> for StreamMap<K, V>

where K: Hash + Eq + Clone,

fn extend<T>(&mut self, iter: T)

where T: IntoIterator<Item = (K, V)>,

Extends a collection with the contents of an iterator.

fn extend_one(&mut self, item: T)

🔬This is a nightly-only experimental API. (extend_one)

Extends a collection with exactly one element.

fn extend_reserve(&mut self, additional: usize)

🔬This is a nightly-only experimental API. (extend_one)

Reserves capacity in a collection for the given number of additional elements.

impl<K, V> FromIterator<(K, V)> for StreamMap<K, V>

where K: Hash + Eq + Clone,

fn from_iter<T: IntoIterator<Item = (K, V)>>(iter: T) -> Self

Creates a value from an iterator.

impl<K, V> Stream for StreamMap<K, V>

where K: Clone + Unpin + Hash + Eq, V: Stream + Unpin,

type Item = (K, <V 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.