Crate key_stream 

Async key-based message streaming library.

Enables sending messages to multiple receivers, grouped by keys, with automatic cleanup of unused keys. No messages are sent until a key is subscribed to, and keys are automatically removed when all receivers are dropped. Cleanup is performed by a background task, so task switching is required for timely removal. The memory usage of the keys map is optimized by shrinking it when many keys are removed.

The main entry point is KeyStream, created with KeyStream::new. It manages the keys and contains a background task for cleaning up unused keys. The cleanup task is immediately notified when a key is dropped, rather than relying on periodic checks. Use KeyStream::sender to obtain a sender handle for sending messages and subscribing to keys. Use KeySender::send to send messages to a key, and KeySender::subscribe to subscribe to a key and receive a KeyReceiver for that key. KeyReceiver::recv can be used to receive messages for a key, waiting asynchronously until a message is available. KeyReceiver can also be converted into a stream of messages using KeyReceiver::to_async_stream.

Examples

use key_stream::KeyStream;
use tokio;

let key_stream = KeyStream::<i32, String>::new(10);
let sender = key_stream.sender();
let mut receiver = sender.subscribe(1).await;
sender.send(&1, "value".to_string()).await.unwrap();
assert_eq!(receiver.recv().await.unwrap(), "value".to_string());

Key cleanup example

use key_stream::KeyStream;
let key_stream = KeyStream::<i32, String>::new(10);
let sender = key_stream.sender();
let receiver = sender.subscribe(1).await;
assert_eq!(key_stream.n_keys().await, 1);
drop(receiver);
// give the key drop task a chance to run
tokio::task::yield_now().await;
let result = sender
    .send(&1, "value".to_string())
    .await
    .unwrap();
assert_eq!(result, 0);
assert_eq!(key_stream.n_keys().await, 0);

Structs

KeyReceiver

A receiver for messages for a specific key.

KeySender

Handle for sending and subscribing to messages by key.

KeyStream

The main entry point for key-based async message streaming.

Traits

Key

Trait bound for types usable as keys in KeyStream. Must be hashable, comparable, cloneable, thread-safe, and 'static.

Value

Trait bound for types usable as values in KeyStream. Must be cloneable, thread-safe, and 'static.