The file is updated

lib/Cargo.toml

@@ -19,7 +19,7 @@ log = "0.4.17"
 tokio = { version = "1.25.0", optional = true }
 bincode = { version = "2.0.0-rc.3", optional = true }
 async-stream = "0.3.5"
-
+dashmap = "6.0.1"
 [package.metadata.docs.rs]
 features = ["remote", "broadcast", "sink-stream"]
 

lib/src/broadcast.rs

@@ -1,9 +1,8 @@
 use std::future::Future;
 use std::pin::Pin;
+use dashmap::DashMap;
 
-use crate::{SSSD};
-
-use std::collections::HashMap;
+use crate::{SSSD}; // Replace with your actual crate items
 
 use std::sync::{
     atomic::{AtomicUsize, Ordering},
@@ -12,27 +11,23 @@ use std::sync::{
 
 use tokio::{
     sync::{
-        mpsc::{
-            self,
-            error::{SendError},
-            Sender
-        },
-        Mutex
-    }
+        mpsc::{self, error::SendError, Sender},
+    },
 };
 
-
+use dashmap::DashMap;
 
 /// Marker-trait for events
-pub trait Event: Copy + Clone + SSSD {}
-impl <S> Event for S where S: Copy + Clone + SSSD {}
+pub trait Event: Copy + Clone + SSSD {} // Replace SSSD with your actual trait
+
+impl<S> Event for S where S: Copy + Clone + SSSD {}
 
 /// `EventCallback` is a marker-trait for callback which will be stored in subscribers
 /// Auto-implemented
 pub trait EventCallback<E>: Send + Sync + 'static {
     fn call<'a>(&'a self, event: E) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>>
-    where
-        Self: 'a;
+        where
+            Self: 'a;
 }
 
 impl<F, E, Fut> EventCallback<E> for F
@@ -42,47 +37,47 @@ where
     Fut: Future<Output = ()> + Send + 'static,
 {
     fn call<'a>(&'a self, event: E) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>>
-    where
-        Self: 'a,
+        where
+            Self: 'a,
     {
         Box::pin(async move {
             self(event).await;
         })
     }
 }
 
-/// # Event Bus
+// # Event Bus
 ///
-/// `EventBus<E,R>`, where `E` implements `Event` marker trait `R` `implements EventCallback<E>`
+/// `EventBus<E>`, where `E` implements `Event`
 ///
 /// `EventBus` accessible from all actors
 ///
-/// on `new()` Event Bus spawnin background task with loop
+/// on `new()` Event Bus spawning background task with loop
 /// which is checking for events
 ///
 /// ## Fields
 ///
-/// * `subscribers` - HashMap that contains id of subsriber and tuple of (Event, Callback) which is executes on event (Variant of `E`)
+/// * `subscribers` - DashMap that contains id of subscriber and tuple of (Event, Callback) which is executed on event (Variant of `E`)
 ///
-/// * `events`      - Vector of all events
+/// * `tx`      - Sender channel for publishing events
+///
+/// * `counter` - Atomic counter for generating subscriber IDs
 ///
 /// ## Methods
 ///
 /// * `new` - Creating new instance of `EventBus`
 ///
-/// * `publish` - Sends an event to a
+/// * `publish` - Sends an event to all subscribers
 ///
-/// * `subscribe` - Creates record in `subscribers` of enum variant and callback to that variant, returns subscriber's id
-///                 and unwinding event stack, it's promissing that subscribers will react on events which are published
-///                 after subscription only
+/// * `subscribe` - Registers a callback for a specific event type
 ///
-/// * `unsubscribe` - Remove record from `subscribers`
+/// * `unsubscribe` - Removes a subscriber based on subscriber ID
 ///
 pub struct EventBus<E>
 where
     E: Event,
 {
-    subscribers: Arc<Mutex<HashMap<usize, Pin<Box<dyn EventCallback<E>>>>>>,
+    subscribers: Arc<DashMap<usize, Pin<Box<dyn EventCallback<E>>>>>,
     tx: Sender<E>,
     counter: AtomicUsize,
 }
@@ -91,12 +86,11 @@ impl<E> EventBus<E>
 where
     E: Event,
 {
-    // FIXME: Calls function only once for some reason...
     pub fn new() -> Self {
         let (tx, mut rx) = mpsc::channel(1000);
 
         let event_bus = Self {
-            subscribers: Arc::new(Mutex::new(HashMap::new())),
+            subscribers: Arc::new(DashMap::new()),
             tx: tx,
             counter: AtomicUsize::new(0),
         };
@@ -107,13 +101,11 @@ where
         let handle = tokio::runtime::Handle::current();
         {
             let _join_handler = handle.spawn(async move {
-                loop {
-                    while let Some(event) = rx.recv().await {
-                        let subscribers = subs.lock().await;
-                        log::trace!("processing event: {:?}", event);
-                        for (_id, callback) in &*subscribers {
-                            callback.call(event).await;
-                        }
+                while let Some(event) = rx.recv().await {
+                    let subscribers = subs.iter();
+                    log::trace!("processing event: {:?}", event);
+                    for (_id, callback) in subscribers {
+                        callback.call(event).await;
                     }
                 }
             });
@@ -132,29 +124,18 @@ where
     where
         F: EventCallback<E>,
     {
-
         let id = self.counter.fetch_add(1, Ordering::SeqCst);
         log::trace!("New subscriber in event bus. id: {}", id);
 
-        let mut subscribers = self.subscribers.lock().await;
-        subscribers.insert(id, Box::pin(callback));
+        self.subscribers.insert(id, Box::pin(callback));
+
         id
     }
 
     pub async fn unsubscribe(&self, subscriber_id: usize) {
-
         log::trace!("Removed subscriber in event bus. id: {}", subscriber_id);
 
-        let mut subscribers = self.subscribers.lock().await;
-        subscribers.remove(&subscriber_id);
+        self.subscribers.remove(&subscriber_id);
     }
-
 }
-
-// impl<E> Default for EventBus<E>
-//     where E: Event,
-// {
-//     fn default() -> Self {
-//         Self::new()
-//     }
-// }
+// Revie this code