Topic alias and message expiry support in broker

rumqttd/CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- Support for topic alias and message expiry in v5
 
 ### Changed
 - Certificate paths configured in config file are checked during startup and throws a panic if it is not valid.

rumqttd/src/link/bridge.rs

@@ -59,7 +59,7 @@ where
         "Starting bridge with subscription on filter \"{}\"",
         &config.sub_path,
     );
-    let (mut tx, mut rx, _ack) = Link::new(None, &config.name, router_tx, true, None, true)?;
+    let (mut tx, mut rx, _ack) = Link::new(None, &config.name, router_tx, true, None, true, None)?;
 
     'outer: loop {
         let mut network = match network_connect(&config, &config.addr, protocol.clone()).await {

rumqttd/src/link/console.rs

@@ -23,7 +23,8 @@ impl ConsoleLink {
     /// Requires the corresponding Router to be running to complete
     pub fn new(config: ConsoleSettings, router_tx: Sender<(ConnectionId, Event)>) -> ConsoleLink {
         let tx = router_tx.clone();
-        let (link_tx, link_rx, _ack) = Link::new(None, "console", tx, true, None, true).unwrap();
+        let (link_tx, link_rx, _ack) =
+            Link::new(None, "console", tx, true, None, true, None).unwrap();
         let connection_id = link_tx.connection_id;
         ConsoleLink {
             config,

rumqttd/src/link/local.rs

@@ -45,6 +45,7 @@ impl Link {
         clean: bool,
         last_will: Option<LastWill>,
         dynamic_filters: bool,
+        topic_alias_max: u16,
     ) -> (
         Event,
         Arc<Mutex<VecDeque<Packet>>>,
@@ -57,6 +58,7 @@ impl Link {
             clean,
             last_will,
             dynamic_filters,
+            topic_alias_max,
         );
         let incoming = Incoming::new(connection.client_id.to_owned());
         let (outgoing, link_rx) = Outgoing::new(connection.client_id.to_owned());
@@ -80,12 +82,19 @@ impl Link {
         clean: bool,
         last_will: Option<LastWill>,
         dynamic_filters: bool,
+        topic_alias_max: Option<u16>,
     ) -> Result<(LinkTx, LinkRx, Notification), LinkError> {
         // Connect to router
         // Local connections to the router shall have access to all subscriptions
 
-        let (message, i, o, link_rx) =
-            Link::prepare(tenant_id, client_id, clean, last_will, dynamic_filters);
+        let (message, i, o, link_rx) = Link::prepare(
+            tenant_id,
+            client_id,
+            clean,
+            last_will,
+            dynamic_filters,
+            topic_alias_max.unwrap_or(0),
+        );
         router_tx.send((0, message))?;
 
         link_rx.recv()?;
@@ -110,20 +119,27 @@ impl Link {
         clean: bool,
         last_will: Option<LastWill>,
         dynamic_filters: bool,
+        topic_alias_max: Option<u16>,
     ) -> Result<(LinkTx, LinkRx, ConnAck), LinkError> {
         // Connect to router
         // Local connections to the router shall have access to all subscriptions
 
-        let (message, i, o, link_rx) =
-            Link::prepare(tenant_id, client_id, clean, last_will, dynamic_filters);
+        let (message, i, o, link_rx) = Link::prepare(
+            tenant_id,
+            client_id,
+            clean,
+            last_will,
+            dynamic_filters,
+            topic_alias_max.unwrap_or(0),
+        );
         router_tx.send_async((0, message)).await?;
 
         link_rx.recv_async().await?;
         let notification = o.lock().pop_front().unwrap();
         // Right now link identifies failure with dropped rx in router,
         // which is probably ok. We need this here to get id assigned by router
         let (id, ack) = match notification {
-            Notification::DeviceAck(Ack::ConnAck(id, ack)) => (id, ack),
+            Notification::DeviceAck(Ack::ConnAck(id, ack, _)) => (id, ack),
             _message => return Err(LinkError::NotConnectionAck),
         };
 

rumqttd/src/link/remote.rs

@@ -73,12 +73,12 @@ impl<P: Protocol> RemoteLink<P> {
         })
         .await??;
 
-        let (connect, lastwill, login) = match packet {
-            Packet::Connect(connect, _, _lastwill, _, login) => {
+        let (connect, props, lastwill, login) = match packet {
+            Packet::Connect(connect, props, _lastwill, _, login) => {
                 Span::current().record("client_id", &connect.client_id);
 
                 // Ignore last will
-                (connect, None, login)
+                (connect, props, None, login)
             }
             packet => return Err(Error::NotConnectPacket(packet)),
         };
@@ -119,14 +119,18 @@ impl<P: Protocol> RemoteLink<P> {
             return Err(Error::InvalidClientId);
         }
 
+        let topic_alias_max = props.and_then(|p| p.topic_alias_max);
+
         let (link_tx, link_rx, notification) = Link::new(
             tenant_id,
             &client_id,
             router_tx,
             clean_session,
             lastwill,
             dynamic_filters,
+            topic_alias_max,
         )?;
+
         let id = link_rx.id();
         Span::current().record("connection_id", id);
 

rumqttd/src/link/shadow.rs

@@ -78,6 +78,7 @@ impl ShadowLink {
             true,
             None,
             config.dynamic_filters,
+            None,
         )?;
         let connection_id = link_rx.id();
 

rumqttd/src/protocol/mod.rs

@@ -146,7 +146,7 @@ pub struct ConnAck {
     pub code: ConnectReturnCode,
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Default)]
 pub struct ConnAckProperties {
     pub session_expiry_interval: Option<u32>,
     pub receive_max: Option<u16>,
@@ -246,7 +246,7 @@ impl Publish {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Default)]
 pub struct PublishProperties {
     pub payload_format_indicator: Option<u8>,
     pub message_expiry_interval: Option<u32>,

rumqttd/src/router/connection.rs

@@ -1,6 +1,8 @@
+use slab::Slab;
+
 use crate::protocol::LastWill;
 use crate::Filter;
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 
 use super::ConnectionEvents;
 
@@ -22,6 +24,12 @@ pub struct Connection {
     pub last_will: Option<LastWill>,
     /// Connection events
     pub events: ConnectionEvents,
+    /// Topic aliases set by clients
+    pub(crate) topic_aliases: HashMap<u16, Filter>,
+    /// Topic aliases used by broker
+    pub(crate) broker_topic_aliases: Option<HashMap<Filter, u16>>,
+    pub topic_alias_max: u16,
+    pub(crate) used_aliases: Slab<()>,
 }
 
 impl Connection {
@@ -32,6 +40,7 @@ impl Connection {
         clean: bool,
         last_will: Option<LastWill>,
         dynamic_filters: bool,
+        topic_alias_max: u16,
     ) -> Connection {
         // Change client id to -> tenant_id.client_id and derive topic path prefix
         // to validate topics
@@ -44,6 +53,17 @@ impl Connection {
             None => (client_id, None),
         };
 
+        let mut used_aliases = Slab::new();
+        // occupy 0th index as 0 is invalid topic alias
+        assert_eq!(0, used_aliases.insert(()));
+
+        // topic_alias_max is 0, that means client doesn't want to use / support topic alias
+        let broker_topic_aliases = if topic_alias_max == 0 {
+            None
+        } else {
+            Some(HashMap::new())
+        };
+
         Connection {
             client_id,
             tenant_prefix,
@@ -52,6 +72,10 @@ impl Connection {
             subscriptions: HashSet::default(),
             last_will,
             events: ConnectionEvents::default(),
+            topic_aliases: HashMap::new(),
+            broker_topic_aliases,
+            topic_alias_max,
+            used_aliases,
         }
     }
 }

rumqttd/src/router/logs.rs

@@ -3,7 +3,8 @@ use slab::Slab;
 use tracing::trace;
 
 use crate::protocol::{
-    matches, ConnAck, PingResp, PubAck, PubComp, PubRec, PubRel, Publish, SubAck, UnsubAck,
+    matches, ConnAck, ConnAckProperties, PingResp, PubAck, PubComp, PubRec, PubRel, Publish,
+    PublishProperties, SubAck, UnsubAck,
 };
 use crate::router::{DataRequest, FilterIdx, SubscriptionMeter, Waiters};
 use crate::{ConnectionId, Filter, Offset, RouterConfig, Topic};
@@ -12,6 +13,34 @@ use crate::segments::{CommitLog, Position};
 use crate::Storage;
 use std::collections::{HashMap, VecDeque};
 use std::io;
+use std::time::Instant;
+
+type PubWithProp = (Publish, Option<PublishProperties>);
+
+#[derive(Clone)]
+pub struct PublishData {
+    pub publish: Publish,
+    pub properties: Option<PublishProperties>,
+    pub timestamp: Instant,
+}
+
+impl From<PubWithProp> for PublishData {
+    fn from((publish, properties): PubWithProp) -> Self {
+        PublishData {
+            publish,
+            properties,
+            timestamp: Instant::now(),
+        }
+    }
+}
+
+// TODO: remove this from here
+impl Storage for PublishData {
+    fn size(&self) -> usize {
+        let publish = &self.publish;
+        4 + publish.topic.len() + publish.payload.len()
+    }
+}
 
 /// Stores 'device' data and 'actions' data in native commitlog
 /// organized by subscription filter. Device data is replicated
@@ -25,10 +54,10 @@ pub struct DataLog {
     /// Also has waiters used to wake connections/replicator tracker
     /// which are caught up with all the data on 'Filter' and waiting
     /// for new data
-    pub native: Slab<Data<Publish>>,
+    pub native: Slab<Data<PublishData>>,
     /// Map of subscription filter name to filter index
     filter_indexes: HashMap<Filter, FilterIdx>,
-    retained_publishes: HashMap<Topic, Publish>,
+    retained_publishes: HashMap<Topic, PublishData>,
     /// List of filters associated with a topic
     publish_filters: HashMap<Topic, Vec<FilterIdx>>,
 }
@@ -151,7 +180,7 @@ impl DataLog {
         filter_idx: FilterIdx,
         offset: Offset,
         len: u64,
-    ) -> io::Result<(Position, Vec<(Publish, Offset)>)> {
+    ) -> io::Result<(Position, Vec<(PubWithProp, Offset)>)> {
         // unwrap to get index of `self.native` is fine here, because when a new subscribe packet
         // arrives in `Router::handle_device_payload`, it first calls the function
         // `next_native_offset` which creates a new commitlog if one doesn't exist. So any new
@@ -163,12 +192,41 @@ impl DataLog {
         // Encoding this information is important so that calling function
         // has more information on how this method behaves.
         let next = data.log.readv(offset, len, &mut o)?;
+
+        let now = Instant::now();
+        o.retain_mut(|(pubdata, _)| {
+            let Some(PublishProperties {
+                message_expiry_interval: Some(t),
+                ..
+            }) = pubdata.properties.as_mut() else {
+                return true;
+            };
+
+            let time_spent = (now - pubdata.timestamp).as_secs() as u32;
+
+            let is_valid = time_spent < *t;
+
+            // ignore expired messages
+            if is_valid {
+                // set message_expiry_interval to (original value - time spent waiting in server)
+                *t -= time_spent;
+            }
+
+            is_valid
+        });
+
+        // no need to include timestamp when returning
+        let o = o
+            .into_iter()
+            .map(|(pubdata, offset)| ((pubdata.publish, pubdata.properties), offset))
+            .collect();
+
         Ok((next, o))
     }
 
-    pub fn shadow(&mut self, filter: &str) -> Option<Publish> {
+    pub fn shadow(&mut self, filter: &str) -> Option<PubWithProp> {
         let data = self.native.get_mut(*self.filter_indexes.get(filter)?)?;
-        data.log.last()
+        data.log.last().map(|p| (p.publish, p.properties))
     }
 
     /// This method is called when the subscriber has caught up with the commit log. In which case,
@@ -196,8 +254,14 @@ impl DataLog {
         inflight
     }
 
-    pub fn insert_to_retained_publishes(&mut self, publish: Publish, topic: Topic) {
-        self.retained_publishes.insert(topic, publish);
+    pub fn insert_to_retained_publishes(
+        &mut self,
+        publish: Publish,
+        publish_properties: Option<PublishProperties>,
+        topic: Topic,
+    ) {
+        let pub_with_props = (publish, publish_properties);
+        self.retained_publishes.insert(topic, pub_with_props.into());
     }
 
     pub fn remove_from_retained_publishes(&mut self, topic: Topic) {
@@ -285,8 +349,8 @@ impl AckLog {
         }
     }
 
-    pub fn connack(&mut self, id: ConnectionId, ack: ConnAck) {
-        let ack = Ack::ConnAck(id, ack);
+    pub fn connack(&mut self, id: ConnectionId, ack: ConnAck, props: Option<ConnAckProperties>) {
+        let ack = Ack::ConnAck(id, ack, props);
         self.committed.push_back(ack);
     }