Crate navvy

A std::sync::mpsc channel broker that allows storing Sender<T> and Receiver<T> of differing types in a single data structure.

The channels are stored in the broker as a HashMap, storing the Channel itself as a Trait object and the type T alongside it. When a Sender<T> or Receiver<T> is requested for a given channel, the type is checked beforehand to ensure that the type of channel requested matches what is stored for the Channel itself. If it matches, the Trait object is cast to the Channel type and returned to the user.

When a Receiver<T> receives, the Sender<T> may still be present on the ChannelBroker which will lead to blocking behaviour so the [Receiver::try_send] method is appropriate if you want to keep the Sender<T> available for future sends.

Once it is determined that no more sends are required, the Sender<T> can be “consumed” which will return the sender from the channel. This is the “last” sender that can be used for this channel. In cases where you just want to consume the sender and receive from the receiver, the ChannelBroker::consume_and_fetch_receiver method will consume the sender for you, closing it to future requests for a sender.

This will allow a user to create channels of different types and keep them under one single data structure, allowing the senders to be sent to different threads and data returned back to the receivers on the broker which can be read when required.

Safety

The cast of a Trait object to a concrete Channel<T> type is unsafe behaviour but as the type is checked before the cast to ensure it matches what is stored for the channel, we can safely cast this to the Channel<T>. Also, as the receivers are owned by thread in which they are created, we can safely cast to &mut Channel<T> under-the-hood and make modifications to the channel without the need to worry about race-conditions.

Usage

use navvy::ChannelBroker;

fn main() {
    // Let's create a Broker which takes a `usize` as it's key
    let mut broker: ChannelBroker<usize> = ChannelBroker::new();

    // Add a channel of type `String` under ID 1
    broker.add_channel::<String>(1);

    // Add another channel of type `u32` under ID 40
    broker.add_channel::<u32>(40);

    // Let's make some threads
    for _ in 0..100 {
        // Get a sender for the String channel with ID 1
        let string_sender = broker.sender::<String>(1).unwrap();

        // Get a sender for the u32 channel with ID 40
        let u32_sender = broker.sender::<u32>(40).unwrap();

        std::thread::spawn(move || {
            for i in 0..50_000 {
                if i % 2 == 0 {
                    u32_sender.send(i as u32).unwrap();
                } else {
                    string_sender.send(format!("{i} is odd!")).unwrap();
                }
            }
        });
    }

    // Get the string receiver because we care about that first
    // We consume the channel sender as we don't need any more senders
    let string_receiver = broker.consume_and_fetch_receiver::<String>(1).unwrap();
    while let Ok(value) = string_receiver.recv() {
        println!("String Value: {value}");
    }

    // Now we want the u32 values
    // Again, we consume the sender of the channel cause we don't need them anymore
    let u32_receiver = broker.consume_and_fetch_receiver::<u32>(40).unwrap();
    while let Ok(value) = u32_receiver.recv() {
        println!("u32 Value: {value}");
    }

    // Close the channels now that we're done
    broker.close_channel(1);
    broker.close_channel(40);
}

Structs

ChannelBroker

Channel ChannelBroker storing channels of arbitrary types under a key K

Enums

NavvyError

Navvy Error Types