diff options
Diffstat (limited to 'src/transport/mycelium')
| -rw-r--r-- | src/transport/mycelium/mod.rs | 24 |
1 files changed, 14 insertions, 10 deletions
diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs index 2bb399c..5e8dc05 100644 --- a/src/transport/mycelium/mod.rs +++ b/src/transport/mycelium/mod.rs @@ -9,13 +9,17 @@ use url::Url; use crate::transport::wire; use crate::transport::{ - EncodedMessage, HandlerFn, MessageHandler, ParsedMessage, Transport, hand_on_probe, is_probe, + Addressed, EncodedMessage, HandlerFn, MessageHandler, ParsedMessage, Transport, hand_on_probe, + is_probe, }; pub(crate) mod daemon; pub(crate) use daemon::{MyceliumSetup, MyceliumSpawnConfig}; +/// The scheme of this transport's addresses (`transport::paths`). +pub(crate) const SCHEME: &str = "mycelium"; + const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300; /// How long Mycelium tries to deliver a message before dropping it. const DELIVERY_LIMIT: Duration = Duration::from_secs(300); @@ -251,22 +255,22 @@ impl Transport for MyceliumTransport { .map_err(anyhow::Error::from) .and_then(|bytes| Ok(serde_json::from_slice::<serde_json::Value>(&bytes)?)) else { - tracing::debug!(to = %message.peer, "Sent message (couldn't decode response from Mycelium REST API)"); + tracing::debug!(to = %Addressed(SCHEME, &message.peer), "Sent message (couldn't decode response from Mycelium REST API)"); return Ok(()); }; let Ok(outbound) = serde_json::from_slice::<OutboundMessage>(&message.body) else { - tracing::debug!(to = %message.peer, body_len = %message.body.len(), "Sent message (couldn't parse outbound message)"); + tracing::debug!(to = %Addressed(SCHEME, &message.peer), body_len = %message.body.len(), "Sent message (couldn't parse outbound message)"); return Ok(()); }; let Ok(parsed) = decode_payload(&outbound.topic, &outbound.payload) else { - tracing::debug!(to = %message.peer, message = ?outbound, "Sent message (couldn't decode outbound message)"); + tracing::debug!(to = %Addressed(SCHEME, &message.peer), message = ?outbound, "Sent message (couldn't decode outbound message)"); return Ok(()); }; let Some(id) = body.get("id").and_then(|id| id.as_str()) else { - tracing::debug!(to = %message.peer, message = ?parsed, "Sent message (couldn't read id from Mycelium REST API response)"); + tracing::debug!(to = %Addressed(SCHEME, &message.peer), message = ?parsed, "Sent message (couldn't read id from Mycelium REST API response)"); return Ok(()); }; - tracing::debug!(to = %message.peer, message = ?parsed, id = %id, "Sent message"); + tracing::debug!(to = %Addressed(SCHEME, &message.peer), message = ?parsed, id = %id, "Sent message"); } Ok(()) } @@ -297,7 +301,7 @@ impl Transport for MyceliumTransport { match parse_message(&inbound_message) { Ok((sender, parsed)) => { tracing::debug!( - from = %inbound_message.src_pk, + from = %Addressed(SCHEME, &inbound_message.src_pk), message = ?parsed, id = %inbound_message.id, //payload_bytes = raw_bytes.len(), @@ -305,7 +309,7 @@ impl Transport for MyceliumTransport { ); let payload_len = base64_decoded_len(&inbound_message.payload); if is_probe(&parsed) { - hand_on_probe(&handler, sender, parsed, payload_len); + hand_on_probe(SCHEME, &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"); } @@ -317,7 +321,7 @@ impl Transport for MyceliumTransport { } if other_versions.insert(inbound_message.src_pk.clone()) { tracing::warn!( - from = %inbound_message.src_pk, + from = %Addressed(SCHEME, &inbound_message.src_pk), version, ours = wire::VERSION, "Peer speaks another wire version (another build); ignoring its messages" @@ -332,7 +336,7 @@ impl Transport for MyceliumTransport { .and_then(|payload| payload.get(1).copied()); tracing::error!( error = %transport_error, - from = %inbound_message.src_pk, + from = %Addressed(SCHEME, &inbound_message.src_pk), kind = ?kind, payload_bytes = base64_decoded_len(&inbound_message.payload), topic = %inbound_message.topic, |