-
Notifications
You must be signed in to change notification settings - Fork 28
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refactor: Add Gossipsub Message wrapper (#436)
# Description This PR implements the following changes: - [x] Add gosssipsub message wrapper - [x] Rename `store_and_notify` to `publish_and_notify` The message wrapper includes a header with a nonce to force the gossip of duplicate receipts. We will likely expand on the header in future work and make the nonce optional. ## Link to issue Implements #421. ## Type of change - [x] Refactor (non-breaking change that updates existing functionality) ## Test plan (required) We've added a unit test to roundtrip a gossiped message to bytes and back again. We also have a gossip notifications integration test to confirm messages are still sent. --------- Co-authored-by: Zeeshan Lakhani <[email protected]>
- Loading branch information
1 parent
0cd381d
commit 59383e5
Showing
5 changed files
with
205 additions
and
42 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,153 @@ | ||
| use anyhow::{anyhow, Result}; | ||
| use homestar_core::workflow::Nonce; | ||
| use libipld::{self, cbor::DagCborCodec, prelude::Codec, serde::from_ipld, Ipld}; | ||
| use std::collections::BTreeMap; | ||
|
|
||
| const HEADER_KEY: &str = "header"; | ||
| const PAYLOAD_KEY: &str = "payload"; | ||
| const NONCE_KEY: &str = "nonce"; | ||
|
|
||
| #[derive(Debug)] | ||
| pub(crate) struct Message<T> { | ||
| pub(crate) header: Header, | ||
| pub(crate) payload: T, | ||
| } | ||
|
|
||
| impl<T> Message<T> { | ||
| pub(crate) fn new(payload: T) -> Self { | ||
| let header = Header { | ||
| nonce: Nonce::generate(), | ||
| }; | ||
|
|
||
| Self { header, payload } | ||
| } | ||
| } | ||
|
|
||
| impl<T> TryFrom<Message<T>> for Vec<u8> | ||
| where | ||
| Ipld: From<Message<T>> + From<T>, | ||
| { | ||
| type Error = anyhow::Error; | ||
|
|
||
| fn try_from(message: Message<T>) -> Result<Self, Self::Error> { | ||
| let message_ipld = Ipld::from(message); | ||
| DagCborCodec.encode(&message_ipld) | ||
| } | ||
| } | ||
|
|
||
| impl<T> TryFrom<Vec<u8>> for Message<T> | ||
| where | ||
| T: TryFrom<Ipld, Error = anyhow::Error>, | ||
| { | ||
| type Error = anyhow::Error; | ||
|
|
||
| fn try_from(bytes: Vec<u8>) -> Result<Self, Self::Error> { | ||
| let ipld: Ipld = DagCborCodec.decode(&bytes)?; | ||
| ipld.try_into() | ||
| .map_err(|_| anyhow!("Could not convert IPLD to pubsub message.")) | ||
| } | ||
| } | ||
|
|
||
| impl<T> From<Message<T>> for Ipld | ||
| where | ||
| Ipld: From<T>, | ||
| { | ||
| fn from(message: Message<T>) -> Self { | ||
| Ipld::Map(BTreeMap::from([ | ||
| (HEADER_KEY.into(), message.header.into()), | ||
| (PAYLOAD_KEY.into(), message.payload.into()), | ||
| ])) | ||
| } | ||
| } | ||
|
|
||
| impl<T> TryFrom<Ipld> for Message<T> | ||
| where | ||
| T: TryFrom<Ipld, Error = anyhow::Error>, | ||
| { | ||
| type Error = anyhow::Error; | ||
|
|
||
| fn try_from(ipld: Ipld) -> Result<Self, Self::Error> { | ||
| let map = from_ipld::<BTreeMap<String, Ipld>>(ipld)?; | ||
|
|
||
| let header = map | ||
| .get(HEADER_KEY) | ||
| .ok_or_else(|| anyhow!("missing {HEADER_KEY}"))? | ||
| .to_owned() | ||
| .try_into()?; | ||
|
|
||
| let payload = map | ||
| .get(PAYLOAD_KEY) | ||
| .ok_or_else(|| anyhow!("missing {PAYLOAD_KEY}"))? | ||
| .to_owned() | ||
| .try_into()?; | ||
|
|
||
| Ok(Message { header, payload }) | ||
| } | ||
| } | ||
|
|
||
| #[derive(Clone, Debug)] | ||
| pub(crate) struct Header { | ||
| nonce: Nonce, | ||
| } | ||
|
|
||
| impl From<Header> for Ipld { | ||
| fn from(header: Header) -> Self { | ||
| Ipld::Map(BTreeMap::from([( | ||
| NONCE_KEY.into(), | ||
| header.nonce.to_owned().into(), | ||
| )])) | ||
| } | ||
| } | ||
|
|
||
| impl TryFrom<Ipld> for Header { | ||
| type Error = anyhow::Error; | ||
|
|
||
| fn try_from(ipld: Ipld) -> Result<Self, Self::Error> { | ||
| let map = from_ipld::<BTreeMap<String, Ipld>>(ipld)?; | ||
|
|
||
| let nonce = map | ||
| .get(NONCE_KEY) | ||
| .ok_or_else(|| anyhow!("Missing {NONCE_KEY}"))? | ||
| .try_into()?; | ||
|
|
||
| Ok(Header { nonce }) | ||
| } | ||
| } | ||
|
|
||
| impl TryFrom<Header> for Vec<u8> { | ||
| type Error = anyhow::Error; | ||
|
|
||
| fn try_from(header: Header) -> Result<Self, Self::Error> { | ||
| let header_ipld = Ipld::from(header); | ||
| DagCborCodec.encode(&header_ipld) | ||
| } | ||
| } | ||
|
|
||
| impl TryFrom<Vec<u8>> for Header { | ||
| type Error = anyhow::Error; | ||
|
|
||
| fn try_from(bytes: Vec<u8>) -> Result<Self, Self::Error> { | ||
| let ipld: Ipld = DagCborCodec.decode(&bytes)?; | ||
| ipld.try_into() | ||
| } | ||
| } | ||
|
|
||
| #[cfg(test)] | ||
| mod test { | ||
| use super::*; | ||
| use crate::{test_utils, Receipt}; | ||
|
|
||
| #[test] | ||
| fn pubsub_message_rountrip() { | ||
| let (_, receipt) = test_utils::receipt::receipts(); | ||
| let message = Message::new(receipt.clone()); | ||
| let bytes: Vec<u8> = message | ||
| .try_into() | ||
| .expect("Could not serialize message into bytes"); | ||
|
|
||
| let parsed = | ||
| Message::<Receipt>::try_from(bytes).expect("Could not deserialize message from bytes"); | ||
|
|
||
| assert_eq!(receipt, parsed.payload); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters