crossbeam
crossbeam copied to clipboard
question - can i filter messages such that they will arrive to certain recipients only and how can receivers be async without blocking their current thread
I want to create a "Notification center" to facilitate decoupled message passing between parts of my application i.e. similar to https://developer.apple.com/documentation/foundation/nsnotificationcenter
I couldn't find an existing solution (maybe you know of one) so started to implement on my own and hit some challenges.
mainly,
the receiver should be async and currently I am spawning threads like crazy and moving the receiver for one of messages in many cases.
the second issue, I'm creating a lot of channels when I could be multiplexing instead on a shared channel.
the issue though is how to "divert"/"filter" the messages to the correct recipients.
Here is some play code I have in place which partially works but suffers from the caveats I mentioned above
lazy_static! {
static ref NOTIFICATION_CALLBACKS: Mutex<HashMap<String, SenderReceiverPair>> =
Mutex::new(HashMap::new());
}
type SenderReceiverPair = (
crossbeam::channel::Sender<NotificationPayload>,
crossbeam::channel::Receiver<NotificationPayload>,
);
#[derive(Clone, Debug)]
pub struct NotificationPayload {
pub notification_id: String,
pub data: Option<serde_json::Value>,
}
pub fn register(
notification_id: String,
) -> crossbeam::channel::Receiver<NotificationPayload> {
if let Some((_, rec)) = NOTIFICATION_CALLBACKS
.lock()
.unwrap()
.get(notification_id.as_str())
{
return rec.clone(); // return a clone of an existing receiver
}
let (sender, reciever) = crossbeam::channel::unbounded::<NotificationPayload>();
NOTIFICATION_CALLBACKS
.lock()
.unwrap()
.insert(notification_id, (sender, reciever.clone())); // re-use the channel if another observer wants to register
reciever
}
pub fn post_notification(notification_id: &str, data: Option<serde_json::Value>) {
if let Some((sender, _)) = NOTIFICATION_CALLBACKS.lock().unwrap().get(notification_id) {
let _ = sender.send(NotificationPayload {
notification_id: notification_id.to_string(),
data,
});
}
}
// example usage
fn do_something() {
// register for some notification non-blocking style
let receiver = notifications::register(some_id);
std::thread::spawn(move || {
for res in receiver.iter() {
info!("got notified for {:?}", res);
}
// how to kill the reciever ? how to kill the thread? do I really need this one off thread?
});
// continue with more work
...
}
// somewhere else posting notification
fn some_other_module_far_away() {
notifications::post(some_id, some_json_payload_with_is_used_by_recievers);
}
What I want (pseudo)
notifications::register(some_notification_name, some_callback);
notifications::post(some_notification_name, some_notifications_data);