1 files changed, 81 insertions, 55 deletions

diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs
index 4edd1fd..585b5ee 100644
--- a/src/transport/mycelium/mod.rs
+++ b/src/transport/mycelium/mod.rs
@@ -15,6 +15,8 @@ pub mod daemon;
 pub use daemon::{MyceliumSetup, MyceliumSpawnConfig};
 
 const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300;
+/// Polls of the inbox in progress at once.
+const INBOX_POLLS: usize = 2;
 /// How long a request to our Mycelium may take, before what its body adds
 /// (`post_timeout`).
 const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
@@ -264,61 +266,19 @@ impl Transport for MyceliumTransport {
             "Starting Mycelium message consumer"
         );
 
-        loop {
-            tracing::trace!("Polling Mycelium inbox");
-
-            let response = match self
-                .poll_client
-                .get(self.messages_endpoint.clone())
-                .query(&[("timeout", MYCELIUM_POLL_TIMEOUT_SECS)])
-                .query(&[("topic", encode_topic(TOPIC))])
-                .send()
-                .await
-            {
-                Ok(resp) => resp,
-                Err(error) => {
-                    tracing::warn!(error = %error, "Mycelium inbox poll failed");
-                    tokio::time::sleep(Duration::from_millis(250)).await;
-                    continue;
-                }
-            };
-
-            let status = response.status();
-
-            if status == StatusCode::NO_CONTENT {
-                tracing::trace!("Mycelium inbox is empty");
-                tokio::time::sleep(EMPTY_QUEUE_DELAY).await;
-                continue;
-            }
-
-            if !status.is_success() {
-                tracing::warn!(status = %status, "Mycelium inbox returned unsuccessful HTTP status");
-                tokio::time::sleep(Duration::from_millis(500)).await;
-                continue;
-            }
-
-            let raw_bytes = match response.bytes().await {
-                Ok(bytes) => bytes,
-                Err(error) => {
-                    tracing::warn!(error = %error, "Failed to read raw Mycelium response stream");
-                    tokio::time::sleep(Duration::from_millis(100)).await;
-                    continue;
-                }
-            };
-
-            let inbound_message = match serde_json::from_slice::<InboundMessage>(&raw_bytes) {
-                Ok(msg) => msg,
-                Err(error) => {
-                    tracing::error!(
-                        error = %error,
-                        raw_payload = %String::from_utf8_lossy(&raw_bytes),
-                        "Failed to decode inbound Mycelium message structure"
-                    );
-                    tokio::time::sleep(Duration::from_millis(500)).await;
-                    continue;
-                }
-            };
+        // Several polls at once, so one message's HTTP round trip overlaps the next one's
+        // and handling them: the messages still go to the handler one at a time.
+        let (tx, mut rx) = tokio::sync::mpsc::channel(INBOX_POLLS * 4);
+        for _ in 0..INBOX_POLLS {
+            tokio::spawn(poll_inbox(
+                self.poll_client.clone(),
+                self.messages_endpoint.clone(),
+                tx.clone(),
+            ));
+        }
+        drop(tx);
 
+        while let Some(inbound_message) = rx.recv().await {
             match parse_message(&inbound_message) {
                 Ok((sender, parsed)) => {
                     tracing::debug!(
@@ -361,10 +321,10 @@ impl Transport for MyceliumTransport {
                         topic = %inbound_message.topic,
                         "Dropping poisoned packet: parsing failed gracefully"
                     );
-                    // No sleep required: we successfully popped a malformed packet and want to process the next one immediately
                 }
             }
         }
+        Ok(())
     }
 
     async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
@@ -373,6 +333,72 @@ impl Transport for MyceliumTransport {
     }
 }
 
+/// Long-polls the inbox and passes what it pops to `tx` (`start_consumer`).
+async fn poll_inbox(
+    client: reqwest::Client,
+    endpoint: Url,
+    tx: tokio::sync::mpsc::Sender<InboundMessage>,
+) {
+    loop {
+        tracing::trace!("Polling Mycelium inbox");
+
+        let response = match client
+            .get(endpoint.clone())
+            .query(&[("timeout", MYCELIUM_POLL_TIMEOUT_SECS)])
+            .query(&[("topic", encode_topic(TOPIC))])
+            .send()
+            .await
+        {
+            Ok(resp) => resp,
+            Err(error) => {
+                tracing::warn!(error = %error, "Mycelium inbox poll failed");
+                tokio::time::sleep(Duration::from_millis(250)).await;
+                continue;
+            }
+        };
+
+        let status = response.status();
+
+        if status == StatusCode::NO_CONTENT {
+            tracing::trace!("Mycelium inbox is empty");
+            tokio::time::sleep(EMPTY_QUEUE_DELAY).await;
+            continue;
+        }
+
+        if !status.is_success() {
+            tracing::warn!(status = %status, "Mycelium inbox returned unsuccessful HTTP status");
+            tokio::time::sleep(Duration::from_millis(500)).await;
+            continue;
+        }
+
+        let raw_bytes = match response.bytes().await {
+            Ok(bytes) => bytes,
+            Err(error) => {
+                tracing::warn!(error = %error, "Failed to read raw Mycelium response stream");
+                tokio::time::sleep(Duration::from_millis(100)).await;
+                continue;
+            }
+        };
+
+        let inbound_message = match serde_json::from_slice::<InboundMessage>(&raw_bytes) {
+            Ok(msg) => msg,
+            Err(error) => {
+                tracing::error!(
+                    error = %error,
+                    raw_payload = %String::from_utf8_lossy(&raw_bytes),
+                    "Failed to decode inbound Mycelium message structure"
+                );
+                tokio::time::sleep(Duration::from_millis(500)).await;
+                continue;
+            }
+        };
+
+        if tx.send(inbound_message).await.is_err() {
+            return; // the consumer stopped
+        }
+    }
+}
+
 /// A message as the Mycelium REST API takes it (the JSON body of an outbound message).
 /// Also the simulator's wire format (`transport::sim`), so tests go through the same
 /// encoding.