Crate futures_concurrency_dynamic 

Dynamic merge combinator built on top of futures-util’s SelectAll.

This crate provides a wrapper around futures_util::stream::SelectAll that adds dynamic stream management capabilities with a convenient API for adding and managing multiple streams at runtime.

1. Direct API with DynamicMerge

Use DynamicMerge directly when you want a single object that owns both the stream and the ability to push new streams:

use futures_concurrency_dynamic::DynamicMerge;
use futures_util::stream::{self, StreamExt, SelectAll};

let mut merge: DynamicMerge<'_, i32> = SelectAll::new();
merge.push(Box::pin(stream::iter(vec![1, 2, 3])));
merge.push(Box::pin(stream::iter(vec![4, 5, 6])));

while let Some(item) = merge.next().await {
    println!("{}", item);
}

2. Handle-based API with dynamic_merge_with_handle

Use the handle-based API when you need separate mutable references to the stream and the push functionality. This is useful when different parts of your code need to own and mutate each component independently:

use futures_concurrency_dynamic::dynamic_merge_with_handle;
use futures_util::stream::{self, StreamExt};

let (mut stream, mut handle) = dynamic_merge_with_handle::<i32>();

// One task can push streams
let producer = tokio::spawn(async move {
    handle.push(stream::iter(vec![1, 2, 3]));
    handle.push(stream::iter(vec![4, 5, 6]));
});

// Another task can consume from the stream
let consumer = tokio::spawn(async move {
    while let Some(item) = stream.next().await {
        println!("{}", item);
    }
});

producer.await.unwrap();
consumer.await.unwrap();

Implementation Note

This crate is built on top of futures_util::stream::SelectAll, providing a more ergonomic API while leveraging the proven implementation from the futures ecosystem. All streams must be Send + 'static. If you need to work with borrowed streams, consider using futures_util::stream::SelectAll directly.

Re-exports

pub use dynamic_merge::DynamicMerge;

pub use handle::dynamic_merge_with_handle;

pub use handle::DynamicMergeHandle;

pub use handle::DynamicMergeStream;

Modules

dynamic_merge

Dynamic merge combinator using SelectAll from futures-util.

handle

Handle-based API for DynamicMerge that allows separate mutable access.