diff options
Diffstat (limited to 'src/transport/mycelium.rs')
| -rw-r--r-- | src/transport/mycelium.rs | 68 |
1 files changed, 49 insertions, 19 deletions
diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs index 82e6016..efab475 100644 --- a/src/transport/mycelium.rs +++ b/src/transport/mycelium.rs @@ -561,13 +561,7 @@ impl Transport for MyceliumTransport { } fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { - 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, - }) + encode_message(peer, message) } async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { @@ -712,6 +706,26 @@ impl Transport for MyceliumTransport { } } +/// A message as the Mycelium REST API takes it (the JSON body of an outbound message). +/// Also the simulator's wire format (`transport::sim`), so tests go through the same +/// encoding. +pub(crate) fn encode_message(peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { + 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, + }) +} + +/// The message in a body made by [`encode_message`]. +#[cfg(test)] +pub(crate) fn decode_message(body: &[u8]) -> Result<ParsedMessage> { + let outbound: OutboundMessage = serde_json::from_slice(body)?; + Ok(decode_payload(&outbound.topic, &outbound.payload)?) +} + fn encode_topic(topic: &str) -> String { BASE64.encode(topic.as_bytes()) } @@ -802,16 +816,20 @@ fn format_message( struct EpochShadow { generation: u64, id: ByteBuf, + /// The sender's epoch changes in its current run (`SyncEnvelope::epoch_seq`). + #[serde(default)] + seq: u32, } -fn epoch_to_shadow(epoch: &Epoch) -> EpochShadow { +fn epoch_to_shadow(epoch: &Epoch, seq: u32) -> EpochShadow { EpochShadow { generation: epoch.generation, id: ByteBuf::from(epoch.id.to_vec()), + seq, } } -fn epoch_from_shadow(shadow: EpochShadow) -> Result<Epoch, TransportError> { +fn epoch_from_shadow(shadow: &EpochShadow) -> Result<Epoch, TransportError> { if shadow.generation == 0 && shadow.id.is_empty() { return Ok(Epoch::GENESIS); } @@ -850,7 +868,7 @@ fn keepalive_to_bytes(msg: &Keepalive) -> Result<Vec<u8>, rmp_serde::encode::Err rmp_serde::to_vec(&KeepaliveShadow { heads, digest, - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), ready: msg.ready, syncs_with_you: msg.syncs_with_you, }) @@ -880,7 +898,8 @@ fn bytes_to_keepalive(bytes: &[u8]) -> Result<Keepalive, TransportError> { }; Ok(Keepalive { heads, - epoch: epoch_from_shadow(shadow.epoch)?, + epoch: epoch_from_shadow(&shadow.epoch)?, + epoch_seq: shadow.epoch.seq, ready: shadow.ready, syncs_with_you: shadow.syncs_with_you, }) @@ -908,6 +927,7 @@ pub struct HelloReplyShadow<'a> { ready: bool, run_id: ByteBuf, away_ms: Option<u64>, + sender_run_id: ByteBuf, } #[derive(Serialize, Deserialize)] @@ -955,7 +975,7 @@ mod automerge_msg_codec { pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> { rmp_serde::to_vec(&HelloRequestShadow { swarm_id: msg.swarm_id.clone(), - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), auth: ByteBuf::from(msg.auth.clone()), ready: msg.ready, run_id: ByteBuf::from(msg.run_id.to_vec()), @@ -965,7 +985,8 @@ pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportErr let shadow: HelloRequestShadow = rmp_serde::from_slice(bytes)?; Ok(HelloRequest { swarm_id: shadow.swarm_id, - epoch: epoch_from_shadow(shadow.epoch)?, + epoch: epoch_from_shadow(&shadow.epoch)?, + epoch_seq: shadow.epoch.seq, auth: shadow.auth.into_vec(), ready: shadow.ready, run_id: run_id_from_shadow(&shadow.run_id)?, @@ -984,11 +1005,12 @@ pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::enco let shadow = HelloReplyShadow { swarm_id: Cow::Borrowed(&msg.swarm_id), peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), auth: ByteBuf::from(msg.auth.clone()), ready: msg.ready, run_id: ByteBuf::from(msg.run_id.to_vec()), away_ms: msg.away_ms, + sender_run_id: ByteBuf::from(msg.sender_run_id.to_vec()), }; rmp_serde::to_vec(&shadow) } @@ -997,11 +1019,13 @@ pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError> Ok(HelloReply { swarm_id: shadow.swarm_id.into_owned(), peers: shadow.peers, - epoch: epoch_from_shadow(shadow.epoch)?, + epoch: epoch_from_shadow(&shadow.epoch)?, + epoch_seq: shadow.epoch.seq, auth: shadow.auth.into_vec(), ready: shadow.ready, run_id: run_id_from_shadow(&shadow.run_id)?, away_ms: shadow.away_ms, + sender_run_id: run_id_from_shadow(&shadow.sender_run_id)?, }) } @@ -1010,7 +1034,7 @@ pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result<Vec<u8>, rmp_serde:: let shadow = SyncEnvelopeShadow { swarm_id: Cow::Borrowed(&msg.swarm_id), message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), ready: msg.ready, }; rmp_serde::to_vec(&shadow) @@ -1019,7 +1043,8 @@ pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportErr let shadow: SyncEnvelopeShadow = rmp_serde::from_slice(bytes)?; Ok(SyncEnvelope { swarm_id: shadow.swarm_id.into_owned(), - epoch: epoch_from_shadow(shadow.epoch)?, + epoch: epoch_from_shadow(&shadow.epoch)?, + epoch_seq: shadow.epoch.seq, message: shadow.message, ready: shadow.ready, }) @@ -1084,7 +1109,7 @@ fn blob_to_bytes(msg: &BlobMessage) -> Result<Vec<u8>, rmp_serde::encode::Error> BlobBody::Missing => BlobBodyShadow::Missing, }; rmp_serde::to_vec(&BlobShadow { - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), id, body, }) @@ -1123,7 +1148,8 @@ fn bytes_to_blob(bytes: &[u8]) -> Result<BlobMessage, TransportError> { BlobBodyShadow::Missing => BlobBody::Missing, }; Ok(BlobMessage { - epoch: epoch_from_shadow(shadow.epoch)?, + epoch: epoch_from_shadow(&shadow.epoch)?, + epoch_seq: shadow.epoch.seq, id, body, }) @@ -1181,6 +1207,7 @@ mod tests { generation: u64::MAX, id: [0xff; 32], }, + epoch_seq: u32::MAX, ready: true, syncs_with_you: true, }; @@ -1203,6 +1230,7 @@ mod tests { let id = BlobId::Change(ChangeHash([9; 32])); let chunk = BlobMessage { epoch, + epoch_seq: 4, id, body: BlobBody::Chunk { index: 5, @@ -1215,12 +1243,14 @@ mod tests { let decoded = bytes_to_blob(&bytes).unwrap(); assert_eq!(decoded.id, id); assert_eq!(decoded.epoch, epoch); + assert_eq!(decoded.epoch_seq, 4); assert!( matches!(decoded.body, BlobBody::Chunk { index: 5, ref data } if data.len() == 64 * 1024) ); let manifest = BlobMessage { epoch, + epoch_seq: 4, id, body: BlobBody::Manifest { size: 100, |