From 11b0b3a3c7b87d902014e0d3d7ab3f14e072c5fa Mon Sep 17 00:00:00 2001
From: James Wilson <james@jsdw.me>
Date: Tue, 10 Aug 2021 10:35:11 +0100
Subject: [PATCH] remove final use of futures::mpsc and replace with flume

---
 backend/telemetry_core/src/aggregator/aggregator_set.rs | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/backend/telemetry_core/src/aggregator/aggregator_set.rs b/backend/telemetry_core/src/aggregator/aggregator_set.rs
index 5fa991b..113eb15 100644
--- a/backend/telemetry_core/src/aggregator/aggregator_set.rs
+++ b/backend/telemetry_core/src/aggregator/aggregator_set.rs
@@ -52,8 +52,11 @@ impl AggregatorSet {
             .map(|a| a.subscribe_shard())
             .collect();
 
+        let (tx, rx) = flume::unbounded::<FromShardWebsocket>();
+        let mut rx = rx.into_stream();
+        let tx = tx.into_sink();
+
         // Send every incoming message to all aggregators.
-        let (tx, mut rx) = futures::channel::mpsc::unbounded::<FromShardWebsocket>();
         tokio::spawn(async move {
             while let Some(msg) = rx.next().await {
                 for conn in &mut conns {