Consolidate `events.map(…).collect()` calls into `Broadcaster`.

This conversion, from an iterator of type-specific events (say, `user::Event` or `message::Event`), into a `Vec<event::Event>`, is prevasive, and it needs to be done each time. Having Broadcaster expose a support method for this cuts down on the repetition, at the cost of a slightly alarming amount of type-system nonsense in `broadcast_from`.

Historical footnote: the internal message structure is a Vec and not an individual message so that bulk operations, like expiring channels and messages, won't disconnect everyone if they happen to dispatch more than sixteen messages (current queue depth limit) at once. We trade allocation and memory pressure for keeping the connections alive. _Most_ event publishing is an iterator of one item, so the Vec allocation is redundant.

6 files changed, 39 insertions, 33 deletions

diff --git a/src/broadcast.rs b/src/broadcast.rs
index 6e1f04d..ee42c08 100644
--- a/src/broadcast.rs
+++ b/src/broadcast.rs
@@ -30,7 +30,7 @@ impl<M> Broadcaster<M>
 where
     M: Clone + Send + std::fmt::Debug + 'static,
 {
-    pub fn broadcast(&self, message: impl Into<M>) {
+    pub fn broadcast(&self, message: M) {
         let tx = self.sender();
 
         // Per the Tokio docs, the returned error is only used to indicate that
@@ -40,7 +40,25 @@ 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 _ = tx.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> {
diff --git a/src/conversation/app.rs b/src/conversation/app.rs
index 5e07292..26886af 100644
--- a/src/conversation/app.rs
+++ b/src/conversation/app.rs
@@ -10,7 +10,7 @@ use super::{
 use crate::{
     clock::DateTime,
     db::{Duplicate as _, NotFound as _},
-    event::{Broadcaster, Event, Sequence, repo::Provider as _},
+    event::{Broadcaster, Sequence, repo::Provider as _},
     message::repo::Provider as _,
     name::{self, Name},
 };
@@ -49,8 +49,7 @@ impl<'a> Conversations<'a> {
 
         tx.commit().await?;
 
-        self.events
-            .broadcast(events.map(Event::from).collect::<Vec<_>>());
+        self.events.broadcast_from(events);
 
         Ok(conversation.as_created())
     }
@@ -105,8 +104,7 @@ impl<'a> Conversations<'a> {
 
         tx.commit().await?;
 
-        self.events
-            .broadcast(events.map(Event::from).collect::<Vec<_>>());
+        self.events.broadcast_from(events);
 
         Ok(())
     }
@@ -134,13 +132,8 @@ impl<'a> Conversations<'a> {
 
         tx.commit().await?;
 
-        self.events.broadcast(
-            events
-                .into_iter()
-                .kmerge_by(Sequence::merge)
-                .map(Event::from)
-                .collect::<Vec<_>>(),
-        );
+        self.events
+            .broadcast_from(events.into_iter().kmerge_by(Sequence::merge));
 
         Ok(())
     }
diff --git a/src/login/app.rs b/src/login/app.rs
index 77d4ac3..e471000 100644
--- a/src/login/app.rs
+++ b/src/login/app.rs
@@ -80,9 +80,10 @@ impl<'a> Logins<'a> {
             tx.tokens().create(&token, &secret).await?;
             tx.commit().await?;
 
-            for event in revoked.into_iter().map(TokenEvent::Revoked) {
-                self.token_events.broadcast(event);
-            }
+            revoked
+                .into_iter()
+                .map(TokenEvent::Revoked)
+                .for_each(|event| self.token_events.broadcast(event));
 
             Ok(secret)
         } else {
diff --git a/src/message/app.rs b/src/message/app.rs
index f0a62d0..647152e 100644
--- a/src/message/app.rs
+++ b/src/message/app.rs
@@ -7,7 +7,7 @@ use crate::{
     clock::DateTime,
     conversation::{self, repo::Provider as _},
     db::NotFound as _,
-    event::{Broadcaster, Event, Sequence, repo::Provider as _},
+    event::{Broadcaster, Sequence, repo::Provider as _},
     login::Login,
     name,
     user::{self, repo::Provider as _},
@@ -62,8 +62,7 @@ impl<'a> Messages<'a> {
 
         tx.commit().await?;
 
-        self.events
-            .broadcast(events.map(Event::from).collect::<Vec<_>>());
+        self.events.broadcast_from(events);
 
         Ok(message.as_sent())
     }
@@ -92,8 +91,7 @@ impl<'a> Messages<'a> {
             tx.messages().record_events(events.clone()).await?;
             tx.commit().await?;
 
-            self.events
-                .broadcast(events.map(Event::from).collect::<Vec<_>>());
+            self.events.broadcast_from(events);
 
             Ok(())
         } else {
@@ -126,13 +124,8 @@ impl<'a> Messages<'a> {
 
         tx.commit().await?;
 
-        self.events.broadcast(
-            events
-                .into_iter()
-                .kmerge_by(Sequence::merge)
-                .map(Event::from)
-                .collect::<Vec<_>>(),
-        );
+        self.events
+            .broadcast_from(events.into_iter().kmerge_by(Sequence::merge));
 
         Ok(())
     }
diff --git a/src/token/app.rs b/src/token/app.rs
index fb5d712..1d68f32 100644
--- a/src/token/app.rs
+++ b/src/token/app.rs
@@ -102,9 +102,10 @@ impl<'a> Tokens<'a> {
         let tokens = tx.tokens().expire(&expire_at).await?;
         tx.commit().await?;
 
-        for event in tokens.into_iter().map(TokenEvent::Revoked) {
-            self.token_events.broadcast(event);
-        }
+        tokens
+            .into_iter()
+            .map(TokenEvent::Revoked)
+            .for_each(|event| self.token_events.broadcast(event));
 
         Ok(())
     }
diff --git a/src/user/create.rs b/src/user/create.rs
index 21c61d1..d6656e5 100644
--- a/src/user/create.rs
+++ b/src/user/create.rs
@@ -99,7 +99,7 @@ where
             login: _,
         } = self;
 
-        events.broadcast(user_events.into_iter().collect::<Vec<_>>());
+        events.broadcast_from(user_events);
     }
 
     pub fn login(&self) -> &Login {