Flumify everything

backend/telemetry_core/src/aggregator/aggregator_set.rs

@@ -1,7 +1,7 @@
 use super::aggregator::{Aggregator, AggregatorOpts};
 use super::inner_loop;
 use common::EitherSink;
-use futures::{Sink, SinkExt, StreamExt};
+use futures::{Sink, SinkExt};
 use inner_loop::FromShardWebsocket;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
@@ -53,12 +53,10 @@ impl AggregatorSet {
             .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.
         tokio::spawn(async move {
-            while let Some(msg) = rx.next().await {
+            while let Ok(msg) = rx.recv_async().await {
                 for conn in &mut conns {
                     // Unbounded channel under the hood, so this await
                     // shouldn't ever need to yield.
@@ -70,7 +68,7 @@ impl AggregatorSet {
             }
         });
 
-        EitherSink::b(tx.sink_map_err(|e| anyhow::anyhow!("{}", e)))
+        EitherSink::b(tx.into_sink().sink_map_err(|e| anyhow::anyhow!("{}", e)))
     }
 
     /// Return a sink that a feed can send messages into to be handled by a single aggregator.

backend/telemetry_core/src/aggregator/inner_loop.rs

@@ -25,7 +25,6 @@ use common::{
     node_types::BlockHash,
     time,
 };
-use futures::StreamExt;
 use std::collections::{HashMap, HashSet};
 use std::{
     net::{IpAddr, Ipv4Addr},
@@ -184,8 +183,7 @@ impl InnerLoop {
         // check the length of the queue below to decide whether or not to
         // pass the message on to this.
         tokio::spawn(async move {
-            let mut metered_rx = metered_rx.into_stream();
-            while let Some(msg) = metered_rx.next().await {
+            while let Ok(msg) = metered_rx.recv_async().await {
                 match msg {
                     ToAggregator::FromFeedWebsocket(feed_conn_id, msg) => {
                         self.handle_from_feed(feed_conn_id, msg)
@@ -215,8 +213,7 @@ impl InnerLoop {
                 });
         });
 
-        let mut rx_from_external = rx_from_external.into_stream();
-        while let Some(msg) = rx_from_external.next().await {
+        while let Ok(msg) = rx_from_external.recv_async().await {
             // ignore node updates if we have too many messages to handle, in an attempt
             // to reduce the queue length back to something reasonable, lest it get out of
             // control and start consuming a load of memory.

backend/telemetry_core/src/find_location.rs

@@ -17,7 +17,7 @@
 use std::net::Ipv4Addr;
 use std::sync::Arc;
 
-use futures::{Sink, SinkExt, StreamExt};
+use futures::{Sink, SinkExt};
 use parking_lot::RwLock;
 use rustc_hash::FxHashMap;
 use serde::Deserialize;
@@ -36,7 +36,6 @@ where
     Id: Clone + Send + 'static,
 {
     let (tx, rx) = flume::unbounded();
-    let mut rx = rx.into_stream();
 
     // cache entries
     let mut cache: FxHashMap<Ipv4Addr, Option<Arc<NodeLocation>>> = FxHashMap::default();
@@ -61,7 +60,7 @@ where
         let semaphore = Arc::new(Semaphore::new(4));
 
         loop {
-            while let Some((id, ip_address)) = rx.next().await {
+            while let Ok((id, ip_address)) = rx.recv_async().await {
                 let permit = semaphore.clone().acquire_owned().await.unwrap();
                 let mut response_chan = response_chan.clone();
                 let locator = locator.clone();