Address feedback from Niklas

README.md

@@ -129,7 +129,7 @@ If you'd like to get things runing manually using Docker, you can do the followi
        substrate-telemetry-frontend
    ```
 
-   **NOTE:** Here we used `SUBSTRATE_TELEMETRY_URL=ws://localhost:8000/feed`. This will work if you test with everything running locally on your machine but NOT if your backend runs on a remote server. Keep in mind that the frontend docker image is serving a static site running your browser. The `SUBSTRATE_TELEMETRY_URL` is the WebSocket url that your browser will use to reach the backend. Say your backend runs on a remore server at `192.168.0.100`, you will need to set the IP/url accordingly in `SUBSTRATE_TELEMETRY_URL`.
+   **NOTE:** Here we used `SUBSTRATE_TELEMETRY_URL=ws://localhost:8000/feed`. This will work if you test with everything running locally on your machine but NOT if your backend runs on a remote server. Keep in mind that the frontend docker image is serving a static site running your browser. The `SUBSTRATE_TELEMETRY_URL` is the WebSocket url that your browser will use to reach the backend. Say your backend runs on a remote server at `foo.example.com`, you will need to set the IP/url accordingly in `SUBSTRATE_TELEMETRY_URL` (in this case, to `ws://foo.example.com/feed`).
 
 With these running, you'll be able to navigate to [http://localhost:3000](http://localhost:3000) to view the UI. If you'd like to connect a node and have it send telemetry to your running shard, you can run the following:
 

backend/common/src/byte_size.rs

@@ -23,7 +23,8 @@ impl ByteSize {
     pub fn new(bytes: usize) -> ByteSize {
         ByteSize(bytes)
     }
-    pub fn into_bytes(self) -> usize {
+    /// Return the number of bytes stored within.
+    pub fn num_bytes(self) -> usize {
         self.0
     }
 }
@@ -101,7 +102,7 @@ mod test {
 
         for (s, expected) in cases {
             let b: ByteSize = s.parse().unwrap();
-            assert_eq!(b.into_bytes(), expected);
+            assert_eq!(b.num_bytes(), expected);
         }
     }
 }

backend/common/src/either_sink.rs

@@ -11,7 +11,7 @@ pin_project! {
 
 /// A simple enum that delegates implementation to one of
 /// the two possible sinks contained within.
-impl <A, B> EitherSink<A, B> {
+impl<A, B> EitherSink<A, B> {
     pub fn a(val: A) -> Self {
         EitherSink::A { inner: val }
     }
@@ -20,38 +20,47 @@ impl <A, B> EitherSink<A, B> {
     }
 }
 
-impl <Item, Error, A, B> Sink<Item> for EitherSink<A, B>
+impl<Item, Error, A, B> Sink<Item> for EitherSink<A, B>
 where
     A: Sink<Item, Error = Error>,
-    B: Sink<Item, Error = Error>
+    B: Sink<Item, Error = Error>,
 {
     type Error = Error;
 
-    fn poll_ready(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll<Result<(), Self::Error>> {
+    fn poll_ready(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), Self::Error>> {
         match self.project() {
-            EitherSinkInner::A{ inner } => inner.poll_ready(cx),
-            EitherSinkInner::B{ inner } => inner.poll_ready(cx)
+            EitherSinkInner::A { inner } => inner.poll_ready(cx),
+            EitherSinkInner::B { inner } => inner.poll_ready(cx),
         }
     }
 
     fn start_send(self: std::pin::Pin<&mut Self>, item: Item) -> Result<(), Self::Error> {
         match self.project() {
-            EitherSinkInner::A{ inner } => inner.start_send(item),
-            EitherSinkInner::B{ inner } => inner.start_send(item)
+            EitherSinkInner::A { inner } => inner.start_send(item),
+            EitherSinkInner::B { inner } => inner.start_send(item),
         }
     }
 
-    fn poll_flush(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll<Result<(), Self::Error>> {
+    fn poll_flush(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), Self::Error>> {
         match self.project() {
-            EitherSinkInner::A{ inner } => inner.poll_flush(cx),
-            EitherSinkInner::B{ inner } => inner.poll_flush(cx)
+            EitherSinkInner::A { inner } => inner.poll_flush(cx),
+            EitherSinkInner::B { inner } => inner.poll_flush(cx),
         }
     }
 
-    fn poll_close(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll<Result<(), Self::Error>> {
+    fn poll_close(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), Self::Error>> {
         match self.project() {
-            EitherSinkInner::A{ inner } => inner.poll_close(cx),
-            EitherSinkInner::B{ inner } => inner.poll_close(cx)
+            EitherSinkInner::A { inner } => inner.poll_close(cx),
+            EitherSinkInner::B { inner } => inner.poll_close(cx),
         }
     }
 }

backend/common/src/http_utils.rs

@@ -153,7 +153,7 @@ fn header_contains_value(
     pub fn trim(x: &[u8]) -> &[u8] {
         let from = match x.iter().position(|x| !x.is_ascii_whitespace()) {
             Some(i) => i,
-            None => return &x[0..0],
+            None => return &[],
         };
         let to = x.iter().rposition(|x| !x.is_ascii_whitespace()).unwrap();
         &x[from..=to]

backend/common/src/lib.rs

@@ -27,15 +27,15 @@ pub mod ws_client;
 
 mod assign_id;
 mod dense_map;
+mod either_sink;
 mod mean_list;
 mod most_seen;
 mod num_stats;
-mod either_sink;
 
 // Export a bunch of common bits at the top level for ease of import:
 pub use assign_id::AssignId;
 pub use dense_map::DenseMap;
+pub use either_sink::EitherSink;
 pub use mean_list::MeanList;
 pub use most_seen::MostSeen;
 pub use num_stats::NumStats;
-pub use either_sink::EitherSink;

backend/common/src/most_seen.rs

@@ -28,6 +28,8 @@ pub struct MostSeen<T> {
 
 impl<T: Default> Default for MostSeen<T> {
     fn default() -> Self {
+        // This sets the "most seen item" to the default value for the type,
+        // and notes that nobody has actually seen it yet (current_count is 0).
         Self {
             current_best: T::default(),
             current_count: 0,
@@ -38,6 +40,9 @@ impl<T: Default> Default for MostSeen<T> {
 
 impl<T> MostSeen<T> {
     pub fn new(item: T) -> Self {
+        // This starts us off with an item that we've seen. This item is set as
+        // the "most seen item" and the current_count is set to 1, as we've seen it
+        // once by virtue of providing it here.
         Self {
             current_best: item,
             current_count: 1,

backend/common/src/node_types.rs

@@ -26,7 +26,7 @@ pub type BlockNumber = u64;
 pub type Timestamp = u64;
 pub use primitive_types::H256 as BlockHash;
 
-///
+/// Basic node details.
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct NodeDetails {
     pub chain: Box<str>,
@@ -38,13 +38,22 @@ pub struct NodeDetails {
     pub startup_time: Option<Box<str>>,
 }
 
-///
+/// A couple of node statistics.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
 pub struct NodeStats {
     pub peers: u64,
     pub txcount: u64,
 }
 
+// # A note about serialization/deserialization of types in this file:
+//
+// Some of the types here are sent to UI feeds. In an effort to keep the
+// amount of bytes sent to a minimum, we have written custom serializers
+// for those types.
+//
+// For testing purposes, it's useful to be able to deserialize from some
+// of these types so that we can test message feed things, so custom
+// deserializers exist to undo the work of the custom serializers.
 impl Serialize for NodeStats {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where
@@ -67,7 +76,7 @@ impl<'de> Deserialize<'de> for NodeStats {
     }
 }
 
-///
+/// Node IO details.
 #[derive(Default)]
 pub struct NodeIO {
     pub used_state_cache_size: MeanList<f32>,
@@ -85,7 +94,7 @@ impl Serialize for NodeIO {
     }
 }
 
-///
+/// Concise block details
 #[derive(Deserialize, Serialize, Debug, Clone, Copy, PartialEq)]
 pub struct Block {
     pub hash: BlockHash,
@@ -101,7 +110,7 @@ impl Block {
     }
 }
 
-///
+/// Node hardware details.
 #[derive(Default)]
 pub struct NodeHardware {
     /// Upload uses means
@@ -126,7 +135,7 @@ impl Serialize for NodeHardware {
     }
 }
 
-///
+/// Node location details
 #[derive(Debug, Clone, PartialEq)]
 pub struct NodeLocation {
     pub latitude: f32,
@@ -161,7 +170,7 @@ impl<'de> Deserialize<'de> for NodeLocation {
     }
 }
 
-///
+/// Verbose block details
 #[derive(Debug, Clone, Copy, PartialEq)]
 pub struct BlockDetails {
     pub block: Block,

backend/common/src/ready_chunks_all.rs

@@ -72,7 +72,7 @@ impl<St: Stream> Stream for ReadyChunksAll<St> {
                     return if this.items.is_empty() {
                         Poll::Pending
                     } else {
-                        Poll::Ready(Some(mem::replace(this.items, Vec::new())))
+                        Poll::Ready(Some(mem::take(this.items)))
                     }
                 }
 
@@ -87,7 +87,7 @@ impl<St: Stream> Stream for ReadyChunksAll<St> {
                     let last = if this.items.is_empty() {
                         None
                     } else {
-                        let full_buf = mem::replace(this.items, Vec::new());
+                        let full_buf = mem::take(this.items);
                         Some(full_buf)
                     };
 

backend/common/src/ws_client/connect.rs

@@ -36,8 +36,8 @@ pub type RawReceiver =
 /// A websocket connection. From this, we can either expose the raw connection
 /// or expose a cancel-safe interface to it.
 pub struct Connection {
-    tx: soketto::connection::Sender<tokio_util::compat::Compat<tokio::net::TcpStream>>,
-    rx: soketto::connection::Receiver<tokio_util::compat::Compat<tokio::net::TcpStream>>,
+    tx: RawSender,
+    rx: RawReceiver,
 }
 
 impl Connection {

backend/telemetry_core/benches/subscribe.rs

@@ -78,7 +78,7 @@ pub fn benchmark_subscribe_speed(c: &mut Criterion) {
 
                 // Give those messages a chance to be handled. This, of course,
                 // assumes that those messages _can_ be handled this quickly. If not,
-                // we'll start to skew benchmark results with the "time taklen to add node".
+                // we'll start to skew benchmark results with the "time taken to add node".
                 tokio::time::sleep(Duration::from_millis(250)).await;
 
                 // Start a bunch of feeds:

backend/telemetry_core/src/aggregator/aggregator_set.rs

@@ -1,10 +1,10 @@
 use super::aggregator::Aggregator;
 use super::inner_loop;
+use common::EitherSink;
 use futures::{Sink, SinkExt, StreamExt};
 use inner_loop::FromShardWebsocket;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
-use common::EitherSink;
 
 #[derive(Clone)]
 pub struct AggregatorSet(Arc<AggregatorSetInner>);
@@ -42,7 +42,7 @@ impl AggregatorSet {
         // if we don't actually need it.
         if self.0.aggregators.len() == 1 {
             let sub = self.0.aggregators[0].subscribe_shard();
-            return EitherSink::a(sub)
+            return EitherSink::a(sub);
         }
 
         let mut conns: Vec<_> = self

backend/telemetry_core/src/aggregator/inner_loop.rs

@@ -438,7 +438,7 @@ impl InnerLoop {
                 // (which is helpful for the UI as it tries to maintain a sorted list of nodes). The chunk
                 // size is the max number of node info we fit into 1 message; smaller messages allow the UI
                 // to react a little faster and not have to wait for a larger update to come in. A chunk size
-                // of 64 means each mesage is ~32k.
+                // of 64 means each message is ~32k.
                 use rayon::prelude::*;
                 let all_feed_messages: Vec<_> = new_chain
                     .nodes_slice()

backend/telemetry_core/src/find_location.rs

@@ -110,7 +110,7 @@ impl Locator {
         // Return location quickly if it's cached:
         let cached_loc = {
             let cache_reader = self.cache.read();
-            cache_reader.get(&ip).map(|o| o.clone())
+            cache_reader.get(&ip).cloned()
         };
         if let Some(loc) = cached_loc {
             return Ok(loc);

backend/telemetry_core/src/main.rs

@@ -213,7 +213,7 @@ where
             // connection anyway.
             let msg_info = tokio::select! {
                 msg_info = ws_recv.receive_data(&mut bytes) => msg_info,
-                _ = &mut recv_closer_rx => { break }
+                _ = &mut recv_closer_rx => break
             };
 
             // Handle the socket closing, or errors receiving the message.

backend/telemetry_shard/src/main.rs

@@ -191,7 +191,7 @@ where
 {
     // Limit the number of bytes based on a rolling total and the incoming bytes per second
     // that has been configured via the CLI opts.
-    let bytes_per_second = bytes_per_second.into_bytes();
+    let bytes_per_second = bytes_per_second.num_bytes();
     let mut rolling_total_bytes = RollingTotalBuilder::new()
         .granularity(Duration::from_secs(1))
         .window_size_multiple(10)