1 files changed, 22 insertions, 0 deletions
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 50841e0..7c31dc6 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -569,6 +569,28 @@ pub(crate) type HandlerFn = dyn Fn(String, ParsedMessage, usize) -> Pin<Box<dyn
     + 'static;
 pub(crate) type MessageHandler = Box<HandlerFn>;
 
+/// Whether `message` is a PING or PONG. The paths layer times them, so a transport's
+/// consumer hands them on as soon as they are read (`hand_on_probe`), not after the
+/// messages read before them are handled (which can take a while: sync).
+pub(crate) fn is_probe(message: &ParsedMessage) -> bool {
+    matches!(message, ParsedMessage::Ping(_) | ParsedMessage::Pong(_))
+}
+
+/// Hands a PING or PONG (`is_probe`) to `handler` in a task of its own.
+pub(crate) fn hand_on_probe(
+    handler: &std::sync::Arc<HandlerFn>,
+    sender: String,
+    message: ParsedMessage,
+    payload_len: usize,
+) {
+    let handler = std::sync::Arc::clone(handler);
+    tokio::spawn(async move {
+        if let Err(error) = handler(sender.clone(), message, payload_len).await {
+            tracing::warn!(peer = %sender, error = %format!("{error:#}"), "Failed to process inbound message");
+        }
+    });
+}
+
 #[async_trait]
 pub(crate) trait Transport: Send + Sync {
     /// This node's identity on the transport (for Mycelium: its public key).