tidyup, add more e2e tests, fix (and test a little) feed msg decoding, supporting bits

backend/test_utils/src/server/channels.rs

@@ -1,3 +1,5 @@
+use std::time::Duration;
+
 use crate::ws_client;
 use futures::{Sink, SinkExt, Stream, StreamExt};
 use crate::feed_message_de::FeedMessage;
@@ -9,6 +11,13 @@ impl From<ws_client::Sender> for ShardSender {
     fn from(c: ws_client::Sender) -> Self { ShardSender(c) }
 }
 
+impl ShardSender {
+    /// Close this connection
+    pub async fn close(&mut self) -> Result<(),ws_client::SendError> {
+        self.0.close().await
+    }
+}
+
 impl Sink<ws_client::Message> for ShardSender {
     type Error = ws_client::SendError;
     fn poll_ready(mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll<Result<(), Self::Error>> {
@@ -97,7 +106,11 @@ impl Stream for FeedReceiver {
 impl FeedReceiver {
     /// Wait for the next set of feed messages to arrive. Returns an error if the connection
     /// is closed, or the messages that come back cannot be properly decoded.
-    pub async fn recv_feed_messages(&mut self) -> Result<Vec<FeedMessage>, anyhow::Error> {
+    ///
+    /// Prefer [`FeedReceiver::recv_feed_messages`]; tests should generally be
+    /// robust in assuming that messages may not all be delivered at once (unless we are
+    /// specifically testing which messages are buffered together).
+    pub async fn recv_feed_messages_once(&mut self) -> Result<Vec<FeedMessage>, anyhow::Error> {
         let msg = self.0
             .next()
             .await
@@ -114,4 +127,28 @@ impl FeedReceiver {
             }
         }
     }
+
+    /// Wait for feed messages to be sent back, building up a list of output messages until
+    /// the channel goes quiet for a short while.
+    pub async fn recv_feed_messages(&mut self) -> Result<Vec<FeedMessage>, anyhow::Error> {
+        // Block as long as needed for messages to start coming in:
+        let mut feed_messages = self.recv_feed_messages_once().await?;
+        // Then, loop a little to make sure we catch any additional messages that are sent soon after:
+        loop {
+            match tokio::time::timeout(Duration::from_millis(250), self.recv_feed_messages_once()).await {
+                // Timeout elapsed; return the messages we have so far
+                Err(_) => {
+                    break Ok(feed_messages);
+                },
+                // Append messages that come back to our vec
+                Ok(Ok(mut msgs)) => {
+                    feed_messages.append(&mut msgs);
+                },
+                // Error came back receiving messages; return it
+                Ok(Err(e)) => {
+                    break Err(e)
+                }
+            }
+        }
+    }
 }

backend/test_utils/src/server/server.rs

@@ -140,9 +140,19 @@ impl Server {
             .await
             .map_err(|e| Error::ErrorObtainingPort(e))?;
 
+        // Attempt to wait until we've received word that the shard is connected to the
+        // core before continuing. If we don't wait for this, the connection may happen
+        // after we've attempted to connect node sockets, and they would be booted and
+        // made to reconnect, which we don't want to deal with in general.
+        let _ = utils::wait_for_line_containing(
+            &mut child_stdout,
+            "Connected to telemetry core",
+            std::time::Duration::from_secs(5)
+        ).await;
+
         // Since we're piping stdout from the child process, we need somewhere for it to go
         // else the process will get stuck when it tries to produce output:
-        utils::drain(child_stdout, tokio::io::stdout());
+        utils::drain(child_stdout, tokio::io::sink());
 
         let shard_uri = format!("http://127.0.0.1:{}/submit", shard_port)
             .parse()
@@ -189,7 +199,7 @@ impl Server {
 
         // Since we're piping stdout from the child process, we need somewhere for it to go
         // else the process will get stuck when it tries to produce output:
-        utils::drain(child_stdout, tokio::io::stdout());
+        utils::drain(child_stdout, tokio::io::sink());
 
         // URI for feeds to connect to the core:
         let feed_uri = format!("http://127.0.0.1:{}/feed", core_port)

backend/test_utils/src/server/utils.rs

@@ -8,19 +8,34 @@ use anyhow::{ anyhow, Context };
 /// with the side benefit that we'll wait for it to start listening before returning. We do this
 /// because we want to allow the kernel to assign ports and so don't specify a port as an arg.
 pub async fn get_port<R: AsyncRead + Unpin>(reader: R) -> Result<u16, anyhow::Error> {
+    let expected_text = "listening on http://127.0.0.1:";
+    wait_for_line_containing(reader, expected_text, Duration::from_secs(30))
+        .await
+        .and_then(|line| {
+            let (_, port_str) = line.rsplit_once(expected_text).unwrap();
+            port_str
+                .trim()
+                .parse()
+                .with_context(|| format!("Could not parse output to port: {}", port_str))
+        })
+}
+
+/// Wait for a line of output containing the text given. Also provide a timeout,
+/// such that if we don't see a new line of output within the timeout we bail out
+/// and return an error.
+pub async fn wait_for_line_containing<R: AsyncRead + Unpin>(reader: R, text: &str, max_wait_between_lines: Duration) -> Result<String, anyhow::Error> {
     let reader = BufReader::new(reader);
     let mut reader_lines = reader.lines();
 
     loop {
         let line = tokio::time::timeout(
-            // This has to accomodate pauses during compilation if the cmd is "cargo run --":
-            Duration::from_secs(30),
+            max_wait_between_lines,
             reader_lines.next_line()
         ).await;
 
         let line = match line {
             // timeout expired; couldn't get port:
-            Err(e) => return Err(anyhow!("Timeout expired waiting to discover port: {}", e)),
+            Err(_) => return Err(anyhow!("Timeout expired waiting for output containing: {}", text)),
             // Something went wrong reading line; bail:
             Ok(Err(e)) => return Err(anyhow!("Could not read line from stdout: {}", e)),
             // No more output; process ended? bail:
@@ -29,15 +44,9 @@ pub async fn get_port<R: AsyncRead + Unpin>(reader: R) -> Result<u16, anyhow::Er
             Ok(Ok(Some(line))) => line
         };
 
-        let (_, port_str) = match line.rsplit_once("listening on http://127.0.0.1:") {
-            Some(m) => m,
-            None => continue
-        };
-
-        return port_str
-            .trim()
-            .parse()
-            .with_context(|| format!("Could not parse output to port: {}", port_str));
+        if line.contains(text) {
+            return Ok(line);
+        }
     }
 }