radicle/node: avoid unnecessary allocations in `Emitter::emit_all`

crates/radicle-protocol/src/service.rs

@@ -1198,16 +1198,14 @@ where
                     rid,
                     updated: updated.clone(),
                 });
-                self.emitter.emit_all(
+                self.emitter
-                    canonical
+                    .emit_all(canonical.into_iter().map(|(refname, target)| {
-                        .into_iter()
+                        Event::CanonicalRefUpdated {
-                        .map(|(refname, target)| Event::CanonicalRefUpdated {
                             rid,
                             refname,
                             target,
-                        })
+                        }
-                        .collect(),
+                    }));
-                );
 
                 // Announce our new inventory if this fetch was a full clone.
                 // Only update and announce inventory for public repositories.

crates/radicle/src/node/events.rs

@@ -228,15 +228,13 @@ impl<T: Clone> Emitter<T> {
     /// Emit a batch of events to subscribers and drop those who can't receive
     /// them.
     /// N.b. subscribers are also dropped if their channel is full.
-    pub fn emit_all(&self, events: Vec<T>) {
+    pub fn emit_all(&self, events: impl IntoIterator<Item = T>) {
         // SAFETY: We deliberately propagate panics from other threads holding the lock.
         #[allow(clippy::unwrap_used)]
-        self.subscribers.lock().unwrap().retain(|s| {
+        let mut subscribers = self.subscribers.lock().unwrap();
-            events
+        for event in events {
-                .clone()
+            subscribers.retain(|s| s.try_send(event.clone()).is_ok());
-                .into_iter()
+        }
-                .all(|event| s.try_send(event).is_ok())
-        });
     }
 
     /// Subscribe to events stream.