diff options
| -rw-r--r-- | src/transport/mycelium/mod.rs | 136 |
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. |