NotifierHub

A Rust crate providing an efficient and asynchronous notification system for broadcasting messages to multiple subscribers. NotifierHub is designed for scenarios that require high performance and concurrent message broadcasting, ensuring non-blocking sends and the ability to wait for completion with timeouts if desired.

Features

• Asynchronous notifications: Broadcast messages to multiple subscribers using asynchronous tasks.
• Subscription management: Easily subscribe and unsubscribe receivers from specific channels.
• Wait-for-send guarantees: Use the WritingHandler to confirm that messages are successfully sent with flexible timeout
• Channel creation waiters: Provide the possibility to wait for the creation of a given channel
• Efficient data handling: Supports sending Arc messages to avoid unnecessary data cloning for large payloads.

Installation

To use Notifier Hub, add the following to your Cargo.toml:

[dependencies]
notifier_hub = "0.1.0"

Getting Started

Here’s a quick example demonstrating how to use NotifierHub to broadcast messages and wait for completion:

Exemple without threads

use notifier_hub::{notifier::NotifierHub, writing_handler::Duration};
    
#[tokio::main]
async fn main() {
    // Create a new NotifierHub
    let mut hub = NotifierHub::new();
    
    // Subscribe to a channel and get a receiver
    let mut receiver1 = hub.subscribe(&"channel1");
    // Subscribe to the same channel and get a receiver
    let mut receiver2 = hub.subscribe(&"channel1");

    // Message to broadcast
    let msg = "Message !";

    // Send the message to all subscribers and get a WritingHandler to track the results
    let handler = hub.clone_send(&msg, &"channel1").unwrap();

    // Wait for up to 100 milliseconds for senders to put the message in the channel buffer
    // This ensures the message is sent successfully or times out.
    handler.wait(Some(Duration::from_millis(100))).await.unwrap();

    assert_eq!(&receiver1.recv().await.unwrap(), &msg);
    assert_eq!(&receiver2.recv().await.unwrap(), &msg);
}

Exemple with threads

use crate::{closable_trait::ClosableMessage, notifier::NotifierHub};
use std::sync::Arc;
use tokio::sync::Mutex;

// This type is going to be sent among the subscribers
#[derive(Clone, Debug)]
enum Message {
    StringMessage(String),
    Number(u32),
    Close,
}

impl ClosableMessage for Message {
    fn get_close_message() -> Self {
        Self::Close
    }
}

// First subscriber
fn subscriber_1(hub: Arc<Mutex<NotifierHub<Message, &'static str>>>) {
    tokio::spawn(async move {
        // Subscribing to "channel1" and "channel2"
        let mut receiver = hub.lock().await
            .subscribe_multiple(&["channel1", "channel2"], 100);
        loop {
            let msg = receiver.recv().await.unwrap();
            match msg {
                Message::StringMessage(s_msg) => {
                    println!("Just received a new message as subscriber_1: {s_msg}")
                }
                Message::Number(n) => println!("Just received a number as subscriber_1: {n}"),
                Message::Close => break,
            }
        }
        // As we are breaking with close, we don't need to unsubscribe
    });
}

// Second subscriber
fn subscriber_2(hub: Arc<Mutex<NotifierHub<Message, &'static str>>>) {
    tokio::spawn(async move {
        // Subscribing only to "channel1"
        let mut receiver = hub.lock().await.subscribe(&"channel1", 100);
        loop {
            let msg = receiver.recv().await.unwrap();
            match msg {
                Message::StringMessage(s_msg) => {
                    println!("Just received a new message as subscriber_2: {s_msg}")
                }
                Message::Number(n) => println!("Just received a number as subscriber_2: {n}"),
                Message::Close => break,
            }
        }
        // As we are breaking with close, we don't need to unsubscribe
    });
}

#[tokio::test]
async fn test_main() {
    // Create a new NotifierHub wrapped in a mutex
    let hub = Arc::new(Mutex::new(NotifierHub::new()));

    let mut creation_waiter = hub.lock().await.get_creation_waiter(&"channel1");
    let mut destruction_waiter = hub.lock().await.get_destruction_waiter(&"channel1");

    // Start subscriber threads
    subscriber_1(hub.clone());
    subscriber_2(hub.clone());

    // Wait for subscriber_1 and subscriber_2 to be ready to receive messages on "channel1"
    creation_waiter.recv().await.unwrap();
    creation_waiter.recv().await.unwrap();

    {
        let mut hub = hub.lock().await;

        let msg1 = Message::StringMessage("Hello!".to_string());
        // Send the message to subscriber_1 and subscriber_2 as they are both subscribed to "channel1"
        hub.clone_send(msg1, &"channel1").unwrap();

        let msg2 = Message::Number(18);
        // Only subscriber_1 will receive this message as it is subscribed to "channel2"
        hub.clone_send(msg2, &"channel2").unwrap();

        let msg3 = Message::StringMessage("Broadcast message!".to_string());
        // Sends msg3 on all channels, so subscriber_1 will receive it twice
        hub.broadcast_clone(msg3);

        // Send a close message to subscriber_1 and subscriber_2
        hub.shutdown_clone(&"channel1").unwrap();
    }

    // Wait for both subscriber_1 and subscriber_2 to unsubscribe
    destruction_waiter.recv().await;
    destruction_waiter.recv().await;
}

Key Concepts

NotifierHub

NotifierHub is the core structure that manages message broadcasting. It holds multiple channels, each identified by an ID, and allows subscribers to register for notifications and also provides some creation waiters able to wait for the creation of a specific channel.

WritingHandler

WritingHandler tracks the status of broadcasted messages. It provides methods to wait for all sends to complete and set optional timeout to avoid indefinite waiting.

Message Types

You can send messages in two ways. You can use the clone_send on NotifierHub that will broadcast the given message to all the subscribers by cloning it. Or the arc_send method using a shared reference to broadcast the data among the subscribers without cloning it.

API Overview

Creating a NotifierHub

let mut hub = NotifierHub::new();

subscribing to channels

let receiver = hub.subscribe(&"channel_id");

Broadcasting Messages through a single channel

Using the cloning broadcast:

let message = "Test message".to_string();
let handler = hub.clone_send(&message, &"channel_id").unwrap();
handler.wait(None).await.unwrap(); // Not necessary, waits for the message to be put in the channel

Using Arc Broadcast (for large messages):

let large_msg = vec![0u8; 10_000_000]; // Large data
let handler = hub.arc_send(large_msg, &"channel_id"); // Will wrap it into an Arc and share it
handler.wait(None).await.unwrap(); // Not necessary, waits for the message to be put in the channel

Unsubscribing

hub.unsubscribe(&"channel_id", &receiver).unwrap();

Getting the State of a Channel

use notifier_hub::notifier::ChannelState;
match hub.channel_state(&"channel_id") {
    ChannelState::Running => println!("Channel is active"),
    ChannelState::Over => println!("Channel has ended"),
    ChannelState::Uninitialised => println!("Channel is uninitialised"),
}

Creation Waiters

You can register waiters to be notified when new subscribers join a channel:

let mut creation_waiter = hub.get_creation_waiter(&"channel1");
let _receiver = hub.subscribe(&"channel1");
assert!(creation_waiter.recv().await.is_some());

Subscribe to Multiple Channels

let receiver = hub.subscribe_multiple(&["channel1", "channel2"]);

Contributing

We welcome contributions! Please open an issue or submit a pull request on GitHub.

License

This project is licensed under the MIT License. See the LICENSE file for details.