1 files changed, 16 insertions, 5 deletions
diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs
index 9075ae8..cc7f25f 100644
--- a/src/transport/mycelium.rs
+++ b/src/transport/mycelium.rs
@@ -559,10 +559,12 @@ impl Transport for MyceliumTransport {
     }
 
     fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
-        let outbound = format_message(peer, message)?;
+        let (outbound, payload_len) = format_message(peer, message)?;
         Ok(EncodedMessage {
             peer: peer.to_owned(),
             body: serde_json::to_vec(&outbound)?.into(),
+            kind: message.kind(),
+            payload_len,
         })
     }
 
@@ -685,7 +687,8 @@ impl Transport for MyceliumTransport {
                         //payload_bytes = raw_bytes.len(),
                         "Received message"
                     );
-                    if let Err(error) = handler(sender, parsed).await {
+                    let payload_len = base64_decoded_len(&inbound_message.payload);
+                    if let Err(error) = handler(sender, parsed, payload_len).await {
                         tracing::warn!(error = %format!("{error:#}"), "Failed to process inbound message");
                     }
                 }
@@ -720,6 +723,12 @@ fn encode_payload(payload: &Vec<u8>) -> String {
     BASE64.encode(payload)
 }
 
+/// Length of what padded base64 `encoded` decodes to.
+fn base64_decoded_len(encoded: &str) -> usize {
+    let padding = encoded.bytes().rev().take_while(|&b| b == b'=').count();
+    (encoded.len() / 4 * 3).saturating_sub(padding)
+}
+
 fn decode_payload(topic_encoded: &str, payload: &str) -> Result<ParsedMessage, TransportError> {
     let topic: &str = &decode_topic(topic_encoded)?;
     let bytes = BASE64.decode(payload)?;
@@ -754,10 +763,11 @@ fn parse_message(message: &InboundMessage) -> Result<(String, ParsedMessage), Tr
     Ok((sender, parsed))
 }
 
+/// The message for Mycelium's API, and the length of its payload.
 fn format_message(
     recipient: &str,
     message: &ParsedMessage,
-) -> Result<OutboundMessage, TransportError> {
+) -> Result<(OutboundMessage, usize), TransportError> {
     let (topic, payload_bytes) = match message {
         ParsedMessage::Hello(req) => (TOPIC_HELLO, hello_request_to_bytes(req)?),
         ParsedMessage::HelloReply(res) => (TOPIC_HELLO_REPLY, hello_reply_to_bytes(res)?),
@@ -765,13 +775,14 @@ fn format_message(
         ParsedMessage::Ping(k) => (TOPIC_PING, keepalive_to_bytes(k)?),
         ParsedMessage::Pong(k) => (TOPIC_PONG, keepalive_to_bytes(k)?),
     };
-    Ok(OutboundMessage {
+    let outbound = OutboundMessage {
         dst: Destination {
             pk: recipient.to_owned(),
         },
         topic: encode_topic(topic),
         payload: encode_payload(&payload_bytes),
-    })
+    };
+    Ok((outbound, payload_bytes.len()))
 }
 
 // =====================================================================