Struct rings_core::message::MessagePayload

pub struct MessagePayload<T> {
    pub data: T,
    pub tx_id: Uuid,
    pub addr: Did,
    pub verification: MessageVerification,
    pub origin_verification: MessageVerification,
    pub relay: MessageRelay,
}

MessagePayload with sequence and verification, contain MessageRelay.

Fields

data: T

tx_id: Uuid

addr: Did

verification: MessageVerification

origin_verification: MessageVerification

relay: MessageRelay

Implementations

impl<T> MessagePayload<T>where
    T: Serialize + DeserializeOwned,

pub fn new(
    data: T,
    session_manager: &SessionManager,
    origin_verification_gen: OriginVerificationGen,
    relay: MessageRelay
) -> Result<Self>

Examples found in repository

src/message/payload.rs (line 122)

    pub fn new_send(
        data: T,
        session_manager: &SessionManager,
        next_hop: Did,
        destination: Did,
    ) -> Result<Self> {
        let relay = MessageRelay::new(
            RelayMethod::SEND,
            vec![session_manager.authorizer()?],
            None,
            Some(next_hop),
            destination,
        );
        Self::new(data, session_manager, OriginVerificationGen::Origin, relay)
    }

    pub fn new_report(
        data: T,
        tx_id: uuid::Uuid,
        session_manager: &SessionManager,
        relay: &MessageRelay,
    ) -> Result<Self> {
        let relay = relay.report()?;
        let mut pl = Self::new(data, session_manager, OriginVerificationGen::Origin, relay)?;
        pl.tx_id = tx_id;
        Ok(pl)
    }

    /// new_direct is A specific new_send, with same next_hop and destination
    /// just like a normal server-client base model.
    pub fn new_direct(data: T, session_manager: &SessionManager, destination: Did) -> Result<Self> {
        Self::new_send(data, session_manager, destination, destination)
    }

    pub fn is_expired(&self) -> bool {
        if self.verification.ttl_ms > MAX_TTL_MS {
            return false;
        }

        if self.origin_verification.ttl_ms > MAX_TTL_MS {
            return false;
        }

        let now = get_epoch_ms();

        if self.verification.ts_ms - TS_OFFSET_TOLERANCE_MS > now {
            return false;
        }

        if self.origin_verification.ts_ms - TS_OFFSET_TOLERANCE_MS > now {
            return false;
        }

        now > self.verification.ts_ms + self.verification.ttl_ms as u128
            && now > self.origin_verification.ts_ms + self.origin_verification.ttl_ms as u128
    }

    pub fn verify(&self) -> bool {
        tracing::debug!("verifying payload: {:?}", self.tx_id);

        if self.is_expired() {
            tracing::warn!("message expired");
            return false;
        }

        self.verification.verify(&self.data) && self.origin_verification.verify(&self.data)
    }

    pub fn origin_session_pubkey(&self) -> Result<PublicKey> {
        self.origin_verification.session_pubkey(&self.data)
    }

    pub fn from_bincode(data: &[u8]) -> Result<Self> {
        bincode::deserialize(data).map_err(Error::BincodeDeserialize)
    }

    pub fn to_bincode(&self) -> Result<Bytes> {
        bincode::serialize(self)
            .map(Bytes::from)
            .map_err(Error::BincodeSerialize)
    }
}

impl<T> Encoder for MessagePayload<T>
where T: Serialize + DeserializeOwned
{
    fn encode(&self) -> Result<Encoded> {
        self.to_bincode()?.encode()
    }
}

impl<T> Decoder for MessagePayload<T>
where T: Serialize + DeserializeOwned
{
    fn from_encoded(encoded: &Encoded) -> Result<Self> {
        let v: Bytes = encoded.decode()?;
        Self::from_bincode(&v)
    }
}

#[cfg_attr(feature = "wasm", async_trait(?Send))]
#[cfg_attr(not(feature = "wasm"), async_trait)]
pub trait PayloadSender<T>
where T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static
{
    fn session_manager(&self) -> &SessionManager;
    async fn do_send_payload(&self, did: Did, payload: MessagePayload<T>) -> Result<()>;

    async fn send_payload(&self, payload: MessagePayload<T>) -> Result<()> {
        if let Some(did) = payload.relay.next_hop {
            self.do_send_payload(did, payload).await
        } else {
            Err(Error::NoNextHop)
        }
    }

    async fn send_message(&self, msg: T, next_hop: Did, destination: Did) -> Result<uuid::Uuid> {
        let payload = MessagePayload::new_send(msg, self.session_manager(), next_hop, destination)?;
        self.send_payload(payload.clone()).await?;
        Ok(payload.tx_id)
    }

    async fn send_direct_message(&self, msg: T, destination: Did) -> Result<uuid::Uuid> {
        let payload = MessagePayload::new_direct(msg, self.session_manager(), destination)?;
        self.send_payload(payload.clone()).await?;
        Ok(payload.tx_id)
    }

    async fn send_report_message(
        &self,
        msg: T,
        tx_id: uuid::Uuid,
        relay: MessageRelay,
    ) -> Result<()> {
        self.send_payload(MessagePayload::new_report(
            msg,
            tx_id,
            self.session_manager(),
            &relay,
        )?)
        .await
    }

    async fn forward_payload(
        &self,
        payload: &MessagePayload<T>,
        relay: MessageRelay,
    ) -> Result<()> {
        self.send_payload(MessagePayload::new(
            payload.data.clone(),
            self.session_manager(),
            OriginVerificationGen::Stick(payload.origin_verification.clone()),
            relay,
        )?)
        .await
    }

src/message/handlers/mod.rs (lines 184-189)

    pub async fn handle_payload(&self, payload: &MessagePayload<Message>) -> Result<()> {
        #[cfg(test)]
        {
            println!("{} got msg {}", self.swarm.did(), &payload.data);
        }
        tracing::debug!("START HANDLE MESSAGE: {} {}", &payload.tx_id, &payload.data);

        self.validate(payload).await?;

        match &payload.data {
            Message::JoinDHT(ref msg) => self.handle(payload, msg).await,
            Message::LeaveDHT(ref msg) => self.handle(payload, msg).await,
            Message::ConnectNodeSend(ref msg) => self.handle(payload, msg).await,
            Message::ConnectNodeReport(ref msg) => self.handle(payload, msg).await,
            Message::AlreadyConnected(ref msg) => self.handle(payload, msg).await,
            Message::FindSuccessorSend(ref msg) => self.handle(payload, msg).await,
            Message::FindSuccessorReport(ref msg) => self.handle(payload, msg).await,
            Message::NotifyPredecessorSend(ref msg) => self.handle(payload, msg).await,
            Message::NotifyPredecessorReport(ref msg) => self.handle(payload, msg).await,
            Message::SearchVNode(ref msg) => self.handle(payload, msg).await,
            Message::FoundVNode(ref msg) => self.handle(payload, msg).await,
            Message::StoreVNode(ref msg) => self.handle(payload, msg).await,
            Message::CustomMessage(ref msg) => self.handle(payload, msg).await,
            Message::MultiCall(ref msg) => {
                for message in msg.messages.iter().cloned() {
                    let payload = MessagePayload::new(
                        message,
                        self.swarm.session_manager(),
                        OriginVerificationGen::Stick(payload.origin_verification.clone()),
                        payload.relay.clone(),
                    )?;
                    self.handle_payload(&payload).await.unwrap_or(());
                }
                Ok(())
            }
            x => Err(Error::MessageHandlerUnsupportedMessageType(format!(
                "{:?}",
                x
            ))),
        }?;

        tracing::debug!("INVOKE CALLBACK {}", &payload.tx_id);
        if let Err(e) = self.invoke_callback(payload).await {
            tracing::warn!("invoke callback error: {}", e);
        }

        tracing::debug!("FINISH HANDLE MESSAGE {}", &payload.tx_id);
        Ok(())
    }

pub fn new_send(
    data: T,
    session_manager: &SessionManager,
    next_hop: Did,
    destination: Did
) -> Result<Self>

Examples found in repository

src/message/payload.rs (line 140)

    pub fn new_direct(data: T, session_manager: &SessionManager, destination: Did) -> Result<Self> {
        Self::new_send(data, session_manager, destination, destination)
    }

    pub fn is_expired(&self) -> bool {
        if self.verification.ttl_ms > MAX_TTL_MS {
            return false;
        }

        if self.origin_verification.ttl_ms > MAX_TTL_MS {
            return false;
        }

        let now = get_epoch_ms();

        if self.verification.ts_ms - TS_OFFSET_TOLERANCE_MS > now {
            return false;
        }

        if self.origin_verification.ts_ms - TS_OFFSET_TOLERANCE_MS > now {
            return false;
        }

        now > self.verification.ts_ms + self.verification.ttl_ms as u128
            && now > self.origin_verification.ts_ms + self.origin_verification.ttl_ms as u128
    }

    pub fn verify(&self) -> bool {
        tracing::debug!("verifying payload: {:?}", self.tx_id);

        if self.is_expired() {
            tracing::warn!("message expired");
            return false;
        }

        self.verification.verify(&self.data) && self.origin_verification.verify(&self.data)
    }

    pub fn origin_session_pubkey(&self) -> Result<PublicKey> {
        self.origin_verification.session_pubkey(&self.data)
    }

    pub fn from_bincode(data: &[u8]) -> Result<Self> {
        bincode::deserialize(data).map_err(Error::BincodeDeserialize)
    }

    pub fn to_bincode(&self) -> Result<Bytes> {
        bincode::serialize(self)
            .map(Bytes::from)
            .map_err(Error::BincodeSerialize)
    }
}

impl<T> Encoder for MessagePayload<T>
where T: Serialize + DeserializeOwned
{
    fn encode(&self) -> Result<Encoded> {
        self.to_bincode()?.encode()
    }
}

impl<T> Decoder for MessagePayload<T>
where T: Serialize + DeserializeOwned
{
    fn from_encoded(encoded: &Encoded) -> Result<Self> {
        let v: Bytes = encoded.decode()?;
        Self::from_bincode(&v)
    }
}

#[cfg_attr(feature = "wasm", async_trait(?Send))]
#[cfg_attr(not(feature = "wasm"), async_trait)]
pub trait PayloadSender<T>
where T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static
{
    fn session_manager(&self) -> &SessionManager;
    async fn do_send_payload(&self, did: Did, payload: MessagePayload<T>) -> Result<()>;

    async fn send_payload(&self, payload: MessagePayload<T>) -> Result<()> {
        if let Some(did) = payload.relay.next_hop {
            self.do_send_payload(did, payload).await
        } else {
            Err(Error::NoNextHop)
        }
    }

    async fn send_message(&self, msg: T, next_hop: Did, destination: Did) -> Result<uuid::Uuid> {
        let payload = MessagePayload::new_send(msg, self.session_manager(), next_hop, destination)?;
        self.send_payload(payload.clone()).await?;
        Ok(payload.tx_id)
    }

pub fn new_report(
    data: T,
    tx_id: Uuid,
    session_manager: &SessionManager,
    relay: &MessageRelay
) -> Result<Self>

Examples found in repository

src/message/payload.rs (lines 243-248)

    async fn send_report_message(
        &self,
        msg: T,
        tx_id: uuid::Uuid,
        relay: MessageRelay,
    ) -> Result<()> {
        self.send_payload(MessagePayload::new_report(
            msg,
            tx_id,
            self.session_manager(),
            &relay,
        )?)
        .await
    }

pub fn new_direct(
    data: T,
    session_manager: &SessionManager,
    destination: Did
) -> Result<Self>

new_direct is A specific new_send, with same next_hop and destination just like a normal server-client base model.

Examples found in repository

src/message/payload.rs (line 232)

    async fn send_direct_message(&self, msg: T, destination: Did) -> Result<uuid::Uuid> {
        let payload = MessagePayload::new_direct(msg, self.session_manager(), destination)?;
        self.send_payload(payload.clone()).await?;
        Ok(payload.tx_id)
    }

src/transports/default/transport.rs (lines 479-483)

    async fn get_handshake_info(
        &self,
        session_manager: &SessionManager,
        kind: RTCSdpType,
        services: HashSet<PeerService>,
    ) -> Result<Encoded> {
        tracing::trace!("prepareing handshake info {:?}", kind);
        let sdp = match kind {
            RTCSdpType::Answer => self.get_answer().await?,
            RTCSdpType::Offer => self.get_offer().await?,
            kind => {
                let mut sdp = self.get_offer().await?;
                sdp.sdp_type = kind;
                sdp
            }
        };
        let local_candidates_json = join_all(
            self.pending_candidates
                .lock()
                .await
                .iter()
                .map(async move |c| c.clone().to_json().unwrap().into()),
        )
        .await;
        if local_candidates_json.is_empty() {
            return Err(Error::FailedOnGatherLocalCandidate);
        }
        let data = TricklePayload {
            sdp: serde_json::to_string(&sdp).unwrap(),
            candidates: local_candidates_json,
            services: services.into_iter().collect_vec(),
        };
        tracing::trace!("prepared handshake info :{:?}", data);
        let resp = MessagePayload::new_direct(
            data,
            session_manager,
            session_manager.authorizer()?.to_owned(), // This is a fake destination
        )?;
        Ok(resp.encode()?)
    }

src/swarm.rs (lines 200-207)

    async fn load_message(
        &self,
        ev: Result<Option<Event>>,
    ) -> Result<Option<MessagePayload<Message>>> {
        let ev = ev?;

        match ev {
            Some(Event::DataChannelMessage(msg)) => {
                let payload = MessagePayload::from_bincode(&msg)?;
                tracing::debug!("load message from channel: {:?}", payload);
                Ok(Some(payload))
            }
            Some(Event::RegisterTransport((did, id))) => {
                // if transport is still pending
                if let Ok(Some(t)) = self.find_pending_transport(id) {
                    tracing::debug!("transport is inside pending list, mov to swarm transports");

                    self.register(did, t).await?;
                    self.pop_pending_transport(id)?;
                }
                match self.get_transport(did) {
                    Some(trans) => {
                        let payload = MessagePayload::new_direct(
                            Message::JoinDHT(message::JoinDHT {
                                did,
                                services: trans.services().await.into_iter().collect(),
                            }),
                            &self.session_manager,
                            self.dht.did,
                        )?;
                        Ok(Some(payload))
                    }
                    None => Err(Error::SwarmMissTransport(did)),
                }
            }
            Some(Event::ConnectClosed((did, uuid))) => {
                if self.pop_pending_transport(uuid).is_ok() {
                    tracing::info!(
                        "[Swarm::ConnectClosed] Pending transport {:?} dropped",
                        uuid
                    );
                };

                if let Some(t) = self.get_transport(did) {
                    if t.id == uuid && self.remove_transport(did).is_some() {
                        tracing::info!("[Swarm::ConnectClosed] transport {:?} closed", uuid);
                        let payload = MessagePayload::new_direct(
                            Message::LeaveDHT(message::LeaveDHT { did }),
                            &self.session_manager,
                            self.dht.did,
                        )?;
                        return Ok(Some(payload));
                    }
                }
                Ok(None)
            }
            None => Ok(None),
        }
    }

pub fn is_expired(&self) -> bool

Examples found in repository

src/message/payload.rs (line 169)

    pub fn verify(&self) -> bool {
        tracing::debug!("verifying payload: {:?}", self.tx_id);

        if self.is_expired() {
            tracing::warn!("message expired");
            return false;
        }

        self.verification.verify(&self.data) && self.origin_verification.verify(&self.data)
    }

pub fn verify(&self) -> bool

Examples found in repository

src/message/handlers/mod.rs (line 213)

    pub async fn listen_once(&self) -> Option<MessagePayload<Message>> {
        if let Some(payload) = self.swarm.poll_message().await {
            if !payload.verify() {
                tracing::error!("Cannot verify msg or it's expired: {:?}", payload);
            }
            if let Err(e) = self.handle_payload(&payload).await {
                tracing::error!("Error in handle_message: {}", e);
                #[cfg(test)]
                {
                    println!("Error in handle_message: {}", e);
                }
            }
            Some(payload)
        } else {
            None
        }
    }
}

#[cfg_attr(feature = "wasm", async_trait(?Send))]
#[cfg_attr(not(feature = "wasm"), async_trait)]
impl PayloadSender<Message> for MessageHandler {
    fn session_manager(&self) -> &SessionManager {
        self.swarm.session_manager()
    }

    async fn do_send_payload(&self, did: Did, payload: MessagePayload<Message>) -> Result<()> {
        self.swarm.do_send_payload(did, payload).await
    }
}

#[cfg(not(feature = "wasm"))]
mod listener {
    use std::sync::Arc;

    use async_trait::async_trait;
    use futures::pin_mut;
    use futures::stream::StreamExt;

    use super::MessageHandler;
    use crate::types::message::MessageListener;

    #[async_trait]
    impl MessageListener for MessageHandler {
        async fn listen(self: Arc<Self>) {
            let payloads = self.swarm.iter_messages().await;
            pin_mut!(payloads);
            while let Some(payload) = payloads.next().await {
                if !payload.verify() {
                    tracing::error!("Cannot verify msg or it's expired: {:?}", payload);
                    continue;
                }
                if let Err(e) = self.handle_payload(&payload).await {
                    tracing::error!("Error in handle_message: {}", e);
                    continue;
                }
            }
        }

src/transports/default/transport.rs (line 490)

    async fn register_remote_info(&self, data: Encoded) -> Result<Did> {
        let data: MessagePayload<TricklePayload> = data.decode()?;
        tracing::trace!("register remote info: {:?}", data);
        match data.verify() {
            true => {
                let sdp = serde_json::from_str::<RTCSessionDescription>(&data.data.sdp)
                    .map_err(Error::Deserialize)?;
                tracing::trace!("setting remote sdp: {:?}", sdp);
                self.set_remote_description(sdp).await?;
                tracing::trace!("setting remote candidate");
                for c in &data.data.candidates {
                    tracing::trace!("add candidates: {:?}", c);
                    if self.add_ice_candidate(c.clone()).await.is_err() {
                        tracing::warn!("failed on add add candidates: {:?}", c.clone());
                    };
                }
                if let Ok(public_key) = data.origin_verification.session.authorizer_pubkey() {
                    let mut pk = self.public_key.write().await;
                    *pk = Some(public_key);
                };
                {
                    let mut services = self.services.write().await;
                    *services = data.data.services.into_iter().collect();
                }
                Ok(data.addr)
            }
            _ => {
                tracing::error!("cannot verify message sig");
                return Err(Error::VerifySignatureFailed);
            }
        }
    }

pub fn origin_session_pubkey(&self) -> Result<PublicKey>

pub fn from_bincode(data: &[u8]) -> Result<Self>

Examples found in repository

src/message/payload.rs (line 205)

    fn from_encoded(encoded: &Encoded) -> Result<Self> {
        let v: Bytes = encoded.decode()?;
        Self::from_bincode(&v)
    }

src/swarm.rs (line 186)

    async fn load_message(
        &self,
        ev: Result<Option<Event>>,
    ) -> Result<Option<MessagePayload<Message>>> {
        let ev = ev?;

        match ev {
            Some(Event::DataChannelMessage(msg)) => {
                let payload = MessagePayload::from_bincode(&msg)?;
                tracing::debug!("load message from channel: {:?}", payload);
                Ok(Some(payload))
            }
            Some(Event::RegisterTransport((did, id))) => {
                // if transport is still pending
                if let Ok(Some(t)) = self.find_pending_transport(id) {
                    tracing::debug!("transport is inside pending list, mov to swarm transports");

                    self.register(did, t).await?;
                    self.pop_pending_transport(id)?;
                }
                match self.get_transport(did) {
                    Some(trans) => {
                        let payload = MessagePayload::new_direct(
                            Message::JoinDHT(message::JoinDHT {
                                did,
                                services: trans.services().await.into_iter().collect(),
                            }),
                            &self.session_manager,
                            self.dht.did,
                        )?;
                        Ok(Some(payload))
                    }
                    None => Err(Error::SwarmMissTransport(did)),
                }
            }
            Some(Event::ConnectClosed((did, uuid))) => {
                if self.pop_pending_transport(uuid).is_ok() {
                    tracing::info!(
                        "[Swarm::ConnectClosed] Pending transport {:?} dropped",
                        uuid
                    );
                };

                if let Some(t) = self.get_transport(did) {
                    if t.id == uuid && self.remove_transport(did).is_some() {
                        tracing::info!("[Swarm::ConnectClosed] transport {:?} closed", uuid);
                        let payload = MessagePayload::new_direct(
                            Message::LeaveDHT(message::LeaveDHT { did }),
                            &self.session_manager,
                            self.dht.did,
                        )?;
                        return Ok(Some(payload));
                    }
                }
                Ok(None)
            }
            None => Ok(None),
        }
    }

pub fn to_bincode(&self) -> Result<Bytes>

Examples found in repository

src/message/payload.rs (line 196)

    fn encode(&self) -> Result<Encoded> {
        self.to_bincode()?.encode()
    }

src/swarm.rs (line 369)

    async fn do_send_payload(&self, did: Did, payload: MessagePayload<T>) -> Result<()> {
        #[cfg(test)]
        {
            println!("+++++++++++++++++++++++++++++++++");
            println!("node {:?}", self.dht.did);
            println!("Sent {:?}", payload.clone());
            println!("node {:?}", payload.relay.next_hop);
            println!("+++++++++++++++++++++++++++++++++");
        }
        let transport = self
            .get_and_check_transport(did)
            .await
            .ok_or(Error::SwarmMissDidInTable(did))?;
        tracing::trace!(
            "SENT {:?}, to node {:?} via transport {:?}",
            payload.clone(),
            payload.relay.next_hop,
            transport.id
        );
        let data = payload.to_bincode()?;
        tracing::info!("send data len: {}", data.len());
        transport.wait_for_data_channel_open().await?;
        transport.send_message(&data).await
    }

Trait Implementations

impl<T: Clone> Clone for MessagePayload<T>

fn clone(&self) -> MessagePayload<T>

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<T: Debug> Debug for MessagePayload<T>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<T> Decoder for MessagePayload<T>where
    T: Serialize + DeserializeOwned,

fn from_encoded(encoded: &Encoded) -> Result<Self>

impl<'de, T> Deserialize<'de> for MessagePayload<T>where
    T: Deserialize<'de>,

fn deserialize<__D>(__deserializer: __D) -> Result<Self, __D::Error>where
    __D: Deserializer<'de>,

Deserialize this value from the given Serde deserializer.

impl<T> Encoder for MessagePayload<T>where
    T: Serialize + DeserializeOwned,

fn encode(&self) -> Result<Encoded>

impl<T: PartialEq> PartialEq<MessagePayload<T>> for MessagePayload<T>

fn eq(&self, other: &MessagePayload<T>) -> bool

This method tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

This method tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.

impl<T> Serialize for MessagePayload<T>where
    T: Serialize,

fn serialize<__S>(&self, __serializer: __S) -> Result<__S::Ok, __S::Error>where
    __S: Serializer,

Serialize this value into the given Serde serializer.

impl<T> TryFrom<MessagePayload<T>> for VirtualNodewhere
    T: Serialize + DeserializeOwned,

type Error = Error

The type returned in the event of a conversion error.

fn try_from(msg: MessagePayload<T>) -> Result<Self>

Performs the conversion.

impl<T: Eq> Eq for MessagePayload<T>

impl<T> StructuralEq for MessagePayload<T>

impl<T> StructuralPartialEq for MessagePayload<T>