1 files changed, 24 insertions, 19 deletions

diff --git a/src/broadcast.rs b/src/broadcast.rs
index 6e1f04d..dae7641 100644
--- a/src/broadcast.rs
+++ b/src/broadcast.rs
@@ -1,16 +1,11 @@
-use std::sync::{Arc, Mutex};
-
 use futures::{Stream, future, stream::StreamExt as _};
 use tokio::sync::broadcast::{Sender, channel};
 use tokio_stream::wrappers::{BroadcastStream, errors::BroadcastStreamRecvError};
 
-// Clones will share the same sender.
+// Clones will share the same channel.
 #[derive(Clone)]
 pub struct Broadcaster<M> {
-    // The use of std::sync::Mutex, and not tokio::sync::Mutex, follows Tokio's
-    // own advice: <https://tokio.rs/tokio/tutorial/shared-state>. Methods that
-    // lock it must be sync.
-    senders: Arc<Mutex<Sender<M>>>,
+    sender: Sender<M>,
 }
 
 impl<M> Default for Broadcaster<M>
@@ -20,9 +15,7 @@ where
     fn default() -> Self {
         let sender = Self::make_sender();
 
-        Self {
-            senders: Arc::new(Mutex::new(sender)),
-        }
+        Self { sender }
     }
 }
 
@@ -30,9 +23,7 @@ impl<M> Broadcaster<M>
 where
     M: Clone + Send + std::fmt::Debug + 'static,
 {
-    pub fn broadcast(&self, message: impl Into<M>) {
-        let tx = self.sender();
-
+    pub fn broadcast(&self, message: M) {
         // Per the Tokio docs, the returned error is only used to indicate that
         // there are no receivers. In this use case, that's fine; a lack of
         // listening consumers (chat clients) when a message is sent isn't an
@@ -40,11 +31,29 @@ where
         //
         // The successful return value, which includes the number of active
         // receivers, also isn't that interesting to us.
-        let _ = tx.send(message.into());
+        let _ = self.sender.send(message);
+    }
+
+    // If `M` is a type that can be obtained from an iterator, such as a `Vec`, and if `I` is an
+    // iterable of items that can be collected into `M`, then this will construct an `M` from the
+    // passed event iterator, converting each element as it goes. This emits one message (as `M`),
+    // containing whatever we collect out of `messages`.
+    //
+    // This is mostly meant for handling synchronized entity events, which tend to be generated as
+    // iterables of domain-specific event types, like `user::Event`, but broadcast as `Vec<event::Event>`
+    // for consumption by outside clients.
+    pub fn broadcast_from<I, E>(&self, messages: I)
+    where
+        I: IntoIterator,
+        M: FromIterator<E>,
+        E: From<I::Item>,
+    {
+        let message = messages.into_iter().map(Into::into).collect();
+        self.broadcast(message);
     }
 
     pub fn subscribe(&self) -> impl Stream<Item = M> + std::fmt::Debug + use<M> {
-        let rx = self.sender().subscribe();
+        let rx = self.sender.subscribe();
 
         BroadcastStream::from(rx).scan((), |(), r| {
             // The following could technically be `r.ok()`, and is exactly
@@ -65,10 +74,6 @@ where
         })
     }
 
-    fn sender(&self) -> Sender<M> {
-        self.senders.lock().unwrap().clone()
-    }
-
     fn make_sender() -> Sender<M> {
         // Queue depth of 16 chosen entirely arbitrarily. Don't read too much
         // into it.