summaryrefslogtreecommitdiffstats
path: root/src/mycelium.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/mycelium.rs')
-rw-r--r--src/mycelium.rs45
1 files changed, 41 insertions, 4 deletions
diff --git a/src/mycelium.rs b/src/mycelium.rs
index 22d49c1..96b9e6d 100644
--- a/src/mycelium.rs
+++ b/src/mycelium.rs
@@ -576,6 +576,21 @@ impl Transport for MyceliumTransport {
}
.into());
}
+
+ // Mycelium's message ID, for GET /api/v1/messages/status/{id} when chasing
+ // a message that never arrived.
+ if tracing::enabled!(tracing::Level::DEBUG) {
+ if let Ok(body) = response
+ .bytes()
+ .await
+ .map_err(anyhow::Error::from)
+ .and_then(|bytes| Ok(serde_json::from_slice::<serde_json::Value>(&bytes)?))
+ {
+ if let Some(id) = body.get("id").and_then(|id| id.as_str()) {
+ tracing::debug!(peer = %message.peer, message_id = id, "Mycelium accepted message");
+ }
+ }
+ }
Ok(())
}
@@ -919,18 +934,25 @@ pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportErr
struct SnapshotRequestShadow {
epoch: EpochShadow,
wanted: EpochShadow,
+ offset: u64,
}
#[derive(Serialize, Deserialize)]
struct SnapshotReplyShadow {
epoch: EpochShadow,
- document: ByteBuf,
+ wanted: EpochShadow,
+ digest: ByteBuf,
+ total: u64,
+ offset: u64,
+ data: ByteBuf,
+ refused: String,
}
fn snapshot_request_to_bytes(msg: &SnapshotRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> {
rmp_serde::to_vec(&SnapshotRequestShadow {
epoch: epoch_to_shadow(&msg.epoch),
wanted: epoch_to_shadow(&msg.wanted),
+ offset: msg.offset,
})
}
@@ -939,22 +961,37 @@ fn bytes_to_snapshot_request(bytes: &[u8]) -> Result<SnapshotRequest, TransportE
Ok(SnapshotRequest {
epoch: epoch_from_shadow(shadow.epoch)?,
wanted: epoch_from_shadow(shadow.wanted)?,
+ offset: shadow.offset,
})
}
fn snapshot_reply_to_bytes(msg: &SnapshotReply) -> Result<Vec<u8>, rmp_serde::encode::Error> {
rmp_serde::to_vec(&SnapshotReplyShadow {
epoch: epoch_to_shadow(&msg.epoch),
- // One copy of the document; replies are rare.
- document: ByteBuf::from(msg.document.clone()),
+ wanted: epoch_to_shadow(&msg.wanted),
+ digest: ByteBuf::from(msg.digest.to_vec()),
+ total: msg.total,
+ offset: msg.offset,
+ data: ByteBuf::from(msg.data.clone()),
+ refused: msg.refused.clone(),
})
}
fn bytes_to_snapshot_reply(bytes: &[u8]) -> Result<SnapshotReply, TransportError> {
let shadow: SnapshotReplyShadow = rmp_serde::from_slice(bytes)?;
+ let digest = shadow
+ .digest
+ .as_slice()
+ .try_into()
+ .map_err(|_| TransportError::InvalidPayload("digest must be 32 bytes".into()))?;
Ok(SnapshotReply {
epoch: epoch_from_shadow(shadow.epoch)?,
- document: shadow.document.into_vec(),
+ wanted: epoch_from_shadow(shadow.wanted)?,
+ digest,
+ total: shadow.total,
+ offset: shadow.offset,
+ data: shadow.data.into_vec(),
+ refused: shadow.refused,
})
}