[v2.0] Introduce Topics

lets/src/address.rs

@@ -22,7 +22,7 @@ use spongos::{
 };
 
 // Local
-use crate::id::Identifier;
+use crate::{id::Identifier, message::Topic};
 
 /// Abstract representation of a Message Address
 ///
@@ -141,12 +141,17 @@ impl AppAddr {
         Self(bytes)
     }
 
-    pub fn gen(identifier: Identifier, app_idx: usize) -> AppAddr {
+    pub fn gen(identifier: Identifier, base_topic: Topic) -> AppAddr {
         let mut addr = [0u8; 40];
         let id_bytes = identifier.as_bytes();
+        // Create spongos to squeeze topic into final 8 bytes
+        let mut s = Spongos::<KeccakF1600>::init();
+        s.absorb(base_topic);
+        s.commit();
+        let squeezed_topic: [u8; 8] = s.squeeze();
         assert_eq!(id_bytes.len(), 32, "identifier must be 32 bytes long");
         addr[..32].copy_from_slice(id_bytes);
-        addr[32..].copy_from_slice(&app_idx.to_be_bytes());
+        addr[32..].copy_from_slice(&squeezed_topic);
         Self::new(addr)
     }
 
@@ -229,10 +234,11 @@ impl MsgId {
         Self(bytes)
     }
 
-    pub fn gen(appaddr: AppAddr, identifier: Identifier, seq_num: usize) -> MsgId {
+    pub fn gen(appaddr: AppAddr, identifier: Identifier, topic: Topic, seq_num: usize) -> MsgId {
         let mut s = Spongos::<KeccakF1600>::init();
         s.absorb(appaddr);
         s.absorb(identifier);
+        s.absorb(topic);
         s.absorb(seq_num.to_be_bytes());
         s.commit();
         s.squeeze()

lets/src/message/hdf.rs

@@ -1,5 +1,4 @@
 use alloc::boxed::Box;
-
 use anyhow::{anyhow, ensure, Result};
 use async_trait::async_trait;
 
@@ -18,12 +17,13 @@ use crate::{
     id::Identifier,
     message::{
         content::{ContentSizeof, ContentUnwrap, ContentWrap},
+        topic::Topic,
         version::{HDF_ID, STREAMS_1_VER, UTF8},
     },
 };
 
 #[non_exhaustive]
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
 #[allow(clippy::upper_case_acronyms)]
 pub struct HDF {
     encoding: u8,
@@ -38,6 +38,7 @@ pub struct HDF {
     pub linked_msg_address: Option<MsgId>,
     pub sequence: usize,
     pub publisher: Identifier,
+    pub topic: Topic,
 }
 
 impl Default for HDF {
@@ -52,12 +53,13 @@ impl Default for HDF {
             linked_msg_address: Default::default(),
             sequence: 0,
             publisher: Default::default(),
+            topic: Default::default(),
         }
     }
 }
 
 impl HDF {
-    pub fn new(message_type: u8, sequence: usize, publisher: Identifier) -> Result<Self> {
+    pub fn new(message_type: u8, sequence: usize, publisher: Identifier, topic: Topic) -> Result<Self> {
         ensure!(
             message_type >> 4 == 0,
             anyhow!(
@@ -75,6 +77,7 @@ impl HDF {
             linked_msg_address: None,
             sequence,
             publisher,
+            topic,
         })
     }
 
@@ -118,6 +121,10 @@ impl HDF {
     pub fn linked_msg_address(&self) -> Option<MsgId> {
         self.linked_msg_address
     }
+
+    pub fn topic(&self) -> &Topic {
+        &self.topic
+    }
 }
 
 #[async_trait(?Send)]
@@ -132,6 +139,7 @@ impl ContentSizeof<HDF> for sizeof::Context {
             .absorb(Uint8::new(hdf.frame_type))?
             .skip(payload_frame_count)?
             .absorb(Maybe::new(hdf.linked_msg_address.as_ref()))?
+            .mask(&hdf.topic)?
             .mask(&hdf.publisher)?
             .skip(Size::new(hdf.sequence))?;
 
@@ -168,6 +176,7 @@ where
             .absorb(Uint8::new(hdf.frame_type))?
             .skip(payload_frame_count)?
             .absorb(Maybe::new(hdf.linked_msg_address.as_ref()))?
+            .mask(&hdf.topic)?
             .mask(&hdf.publisher)?
             .skip(Size::new(hdf.sequence))?;
 
@@ -217,6 +226,7 @@ where
                 anyhow!("first 2 bits of payload-frame-count are reserved"),
             )?
             .absorb(Maybe::new(&mut hdf.linked_msg_address))?
+            .mask(&mut hdf.topic)?
             .mask(&mut hdf.publisher)?
             .skip(&mut seq_num)?;
 
@@ -232,7 +242,6 @@ where
         x[2] = payload_frame_count_bytes[1];
         x[3] = payload_frame_count_bytes[2];
         hdf.payload_frame_count = u32::from_be_bytes(x);
-
         hdf.sequence = seq_num.inner();
 
         Ok(self)

lets/src/message/message.rs

@@ -19,7 +19,7 @@ use super::{
     transport::TransportMessage,
 };
 
-#[derive(Clone, Copy, PartialEq, Eq, Hash, Default, Debug)]
+#[derive(Clone, PartialEq, Eq, Hash, Default, Debug)]
 pub struct Message<Content> {
     header: HDF,
     payload: PCF<Content>,
@@ -41,7 +41,7 @@ impl<Payload> Message<Payload> {
     }
 
     pub fn header(&self) -> HDF {
-        self.header
+        self.header.clone()
     }
 
     pub fn payload(&self) -> &PCF<Payload> {

lets/src/message/mod.rs

@@ -12,6 +12,8 @@ mod version;
 /// Linked Message with header already parsed
 mod preparsed;
 
+pub mod topic;
+
 mod message;
 
 pub use content::{
@@ -22,4 +24,5 @@ pub use hdf::HDF;
 pub use message::Message;
 pub use pcf::PCF;
 pub use preparsed::PreparsedMessage;
+pub use topic::Topic;
 pub use transport::TransportMessage;

lets/src/message/preparsed.rs

@@ -32,7 +32,7 @@ impl<F> PreparsedMessage<F> {
     }
 
     pub fn header(&self) -> HDF {
-        self.header
+        self.header.clone()
     }
 
     pub fn transport_msg(&self) -> &TransportMessage {

lets/src/message/topic.rs

@@ -0,0 +1,81 @@
+use alloc::{
+    string::{String, ToString},
+    vec::Vec,
+};
+use core::{convert::TryFrom, fmt::Formatter};
+use spongos::{
+    ddml::{
+        commands::{sizeof, unwrap, wrap, Mask},
+        io,
+        types::Bytes,
+    },
+    PRP,
+};
+
+#[derive(Clone, PartialEq, Eq, Debug, Default, Hash)]
+pub struct Topic(String);
+
+impl Topic {
+    pub fn new(t: String) -> Self {
+        Self(t)
+    }
+}
+
+impl From<&str> for Topic {
+    fn from(t: &str) -> Self {
+        Self(t.to_string())
+    }
+}
+
+impl TryFrom<&[u8]> for Topic {
+    type Error = anyhow::Error;
+    fn try_from(t: &[u8]) -> Result<Self, Self::Error> {
+        let topic = String::from_utf8(t.to_vec())?;
+        Ok(Topic(topic))
+    }
+}
+
+impl core::fmt::Display for Topic {
+    fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
+        write!(f, "{}", &self.0)
+    }
+}
+
+impl AsRef<[u8]> for Topic {
+    fn as_ref(&self) -> &[u8] {
+        self.0.as_ref()
+    }
+}
+
+impl AsMut<Vec<u8>> for Topic {
+    fn as_mut(&mut self) -> &mut Vec<u8> {
+        unsafe { self.0.as_mut_vec() }
+    }
+}
+
+impl Mask<&Topic> for sizeof::Context {
+    fn mask(&mut self, topic: &Topic) -> anyhow::Result<&mut Self> {
+        self.mask(Bytes::new(topic))
+    }
+}
+
+impl<OS, F> Mask<&Topic> for wrap::Context<OS, F>
+where
+    F: PRP,
+    OS: io::OStream,
+{
+    fn mask(&mut self, topic: &Topic) -> anyhow::Result<&mut Self> {
+        self.mask(Bytes::new(topic))
+    }
+}
+
+impl<IS, F> Mask<&mut Topic> for unwrap::Context<IS, F>
+where
+    F: PRP,
+    IS: io::IStream,
+{
+    fn mask(&mut self, topic: &mut Topic) -> anyhow::Result<&mut Self> {
+        self.mask(Bytes::new(topic.as_mut()))?;
+        Ok(self)
+    }
+}

spongos/src/ddml/commands/unwrap/absorb.rs

@@ -91,7 +91,7 @@ impl<F: PRP, IS: io::IStream> Absorb<Bytes<&mut Vec<u8>>> for Context<IS, F> {
     fn absorb(&mut self, mut bytes: Bytes<&mut Vec<u8>>) -> Result<&mut Self> {
         let mut size = Size::default();
         self.absorb(&mut size)?;
-        bytes.resize(size.inner());
+        bytes.resize(size.inner())?;
         AbsorbContext::new(self).unwrapn(bytes)?;
         Ok(self)
     }

spongos/src/ddml/commands/unwrap/mask.rs

@@ -89,7 +89,7 @@ impl<'a, F: PRP, IS: io::IStream> Mask<Bytes<&'a mut Vec<u8>>> for Context<IS, F
     fn mask(&mut self, mut bytes: Bytes<&'a mut Vec<u8>>) -> Result<&mut Self> {
         let mut size = Size::default();
         self.mask(&mut size)?;
-        bytes.resize(size.inner());
+        bytes.resize(size.inner())?;
         MaskContext::new(self).unwrapn(bytes)?;
         Ok(self)
     }

spongos/src/ddml/commands/unwrap/skip.rs

@@ -86,7 +86,7 @@ impl<'a, F, IS: io::IStream> Skip<Bytes<&'a mut Vec<u8>>> for Context<IS, F> {
     fn skip(&mut self, mut bytes: Bytes<&'a mut Vec<u8>>) -> Result<&mut Self> {
         let mut size = Size::default();
         self.skip(&mut size)?;
-        bytes.resize(size.inner());
+        bytes.resize(size.inner())?;
         SkipContext::new(self).unwrapn(bytes)?;
         Ok(self)
     }

spongos/src/ddml/types/bytes.rs

@@ -1,6 +1,7 @@
 use core::{fmt, iter::FromIterator};
 
 use alloc::{string::String, vec::Vec};
+use anyhow::anyhow;
 
 /// Variable-size array of bytes, the size is not known at compile time and is encoded in trinary
 /// representation.
@@ -60,8 +61,14 @@ where
 }
 
 impl Bytes<&mut Vec<u8>> {
-    pub(crate) fn resize(&mut self, new_size: usize) {
-        self.0.resize(new_size, 0)
+    // Return an error if size exceeds u32 max size to avoid panic
+    // TODO: Remove anyhow error and replace with no_std compatible error handling
+    pub(crate) fn resize(&mut self, new_size: usize) -> anyhow::Result<()> {
+        if new_size > u32::MAX as usize {
+            return Err(anyhow!("{} exceeds maximum available space for Byte resize", new_size));
+        }
+        self.0.resize(new_size, 0);
+        Ok(())
     }
 }