Shared subscriptions with configurable strategy

rumqttd/Cargo.toml

@@ -34,6 +34,7 @@ metrics = "0.21.1"
 metrics-exporter-prometheus = "0.12.1"
 clap = { version = "4.2", features = ["derive"] }
 axum = "0.6.4"
+rand = "0.8.5"
 
 [features]
 default = ["use-rustls"]

rumqttd/rumqttd.toml

@@ -9,6 +9,7 @@ max_connections = 10010
 max_outgoing_packet_count = 200
 max_segment_size = 104857600
 max_segment_count = 10
+# shared_subscriptions_strategy = "random" # "sticky" | "roundrobin" ( default ) | "random"
 # Any filters that match to configured filter will have custom segment size.
     # [router.custom_segment.'/office/+/devices/status']
     # max_segment_size = 102400

rumqttd/src/lib.rs

@@ -38,6 +38,8 @@ pub use link::meters;
 pub use router::{Alert, IncomingMeter, Meter, Notification, OutgoingMeter};
 pub use server::Broker;
 
+use self::router::shared_subs::Strategy;
+
 #[derive(Debug, Default, Serialize, Deserialize, Clone)]
 pub struct Config {
     pub id: usize,
@@ -144,6 +146,9 @@ pub struct RouterConfig {
     pub max_segment_count: usize,
     pub custom_segment: Option<HashMap<String, SegmentConfig>>,
     pub initialized_filters: Option<Vec<Filter>>,
+    // defaults to Round Robin
+    #[serde(default)]
+    pub shared_subscriptions_strategy: Strategy,
 }
 
 #[derive(Debug, Default, Clone, Serialize, Deserialize)]

rumqttd/src/router/logs.rs

@@ -415,6 +415,7 @@ impl AckLog {
 #[cfg(test)]
 mod test {
     use super::DataLog;
+    use crate::router::shared_subs::Strategy;
     use crate::RouterConfig;
 
     #[test]
@@ -426,6 +427,7 @@ mod test {
             max_outgoing_packet_count: 1024,
             custom_segment: None,
             initialized_filters: None,
+            shared_subscriptions_strategy: Strategy::RoundRobin,
         };
         let mut data = DataLog::new(config).unwrap();
         data.next_native_offset("topic/a");
@@ -445,6 +447,7 @@ mod test {
             max_outgoing_packet_count: 1024,
             custom_segment: None,
             initialized_filters: None,
+            shared_subscriptions_strategy: Strategy::RoundRobin,
         };
         let mut data = DataLog::new(config).unwrap();
         data.next_native_offset("+/+");

rumqttd/src/router/mod.rs

@@ -22,6 +22,7 @@ pub mod iobufs;
 mod logs;
 mod routing;
 mod scheduler;
+pub(crate) mod shared_subs;
 mod waiters;
 
 pub use alertlog::Alert;
@@ -191,6 +192,7 @@ pub struct DataRequest {
     pub read_count: usize,
     /// Maximum count of payload buffer per replica
     max_count: usize,
+    pub(crate) group: Option<String>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]