From de75b63c44235fab3547ac2e060e610bff94646a Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 28 Oct 2026 05:47:22 +0100 Subject: don't queue PING/PONG probes --- src/transport/mycelium/mod.rs | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) (limited to 'src/transport/mycelium') diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs index 61aab85..2bb399c 100644 --- a/src/transport/mycelium/mod.rs +++ b/src/transport/mycelium/mod.rs @@ -8,7 +8,9 @@ use thiserror::Error; use url::Url; use crate::transport::wire; -use crate::transport::{EncodedMessage, MessageHandler, ParsedMessage, Transport}; +use crate::transport::{ + EncodedMessage, HandlerFn, MessageHandler, ParsedMessage, Transport, hand_on_probe, is_probe, +}; pub(crate) mod daemon; @@ -289,6 +291,8 @@ impl Transport for MyceliumTransport { } drop(tx); + // PINGs and PONGs aren't handled after the message before them (`is_probe`). + let handler: std::sync::Arc = std::sync::Arc::from(handler); while let Some(inbound_message) = rx.recv().await { match parse_message(&inbound_message) { Ok((sender, parsed)) => { @@ -300,7 +304,9 @@ impl Transport for MyceliumTransport { "Received message" ); let payload_len = base64_decoded_len(&inbound_message.payload); - if let Err(error) = handler(sender, parsed, payload_len).await { + if is_probe(&parsed) { + hand_on_probe(&handler, sender, parsed, payload_len); + } else if let Err(error) = handler(sender, parsed, payload_len).await { tracing::warn!(error = %format!("{error:#}"), "Failed to process inbound message"); } } -- cgit v1.3.1