*: introduce basic infrastructure for pub-sub ibus communication

This commit introduces the `IbusSubscriber` struct, which allows
identification of subscribers in ibus messages and enables receivers
to send notifications back to them.

For convenience, a `subscriber` field has been added to
`IbusChannelsTx`. This field can now be used by any base component
or protocol instance to send subscription messages.

Additionally, a new `Disconnect` ibus message has been
introduced. This message is automatically sent to all base components
by protocol instances when they are shutting down, relieving them
from having to manually undo subscriptions.

Signed-off-by: Renato Westphal <[email protected]>

holo-interface/src/lib.rs

@@ -17,7 +17,7 @@ use holo_northbound::{
     process_northbound_msg,
 };
 use holo_protocol::InstanceShared;
-use holo_utils::ibus::{IbusChannelsTx, IbusReceiver};
+use holo_utils::ibus::{IbusChannelsTx, IbusReceiver, IbusSubscriber};
 use tokio::sync::mpsc;
 use tracing::Instrument;
 
@@ -74,11 +74,12 @@ impl Master {
 
 pub fn start(
     nb_tx: NbProviderSender,
-    ibus_tx: IbusChannelsTx,
+    mut ibus_tx: IbusChannelsTx,
     ibus_rx: IbusReceiver,
     shared: InstanceShared,
 ) -> NbDaemonSender {
     let (nb_daemon_tx, nb_daemon_rx) = mpsc::channel(4);
+    ibus_tx.subscriber = Some(IbusSubscriber::new(ibus_tx.interface.clone()));
 
     tokio::spawn(async move {
         // Initialize netlink socket.

holo-interface/src/northbound/configuration.rs

@@ -446,6 +446,7 @@ impl Provider for Master {
                             ifname,
                             &self.nb_tx,
                             &self.ibus_tx,
+                            ibus_instance_tx.clone(),
                             ibus_instance_rx,
                             Default::default(),
                             self.shared.clone(),

holo-keychain/src/lib.rs

@@ -13,7 +13,7 @@ use holo_northbound::{
     NbDaemonReceiver, NbDaemonSender, NbProviderSender, ProviderBase,
     process_northbound_msg,
 };
-use holo_utils::ibus::{IbusChannelsTx, IbusReceiver};
+use holo_utils::ibus::{IbusChannelsTx, IbusReceiver, IbusSubscriber};
 use holo_utils::keychain::Keychain;
 use tokio::sync::mpsc;
 use tracing::Instrument;
@@ -61,10 +61,11 @@ impl Master {
 
 pub fn start(
     nb_provider_tx: NbProviderSender,
-    ibus_tx: IbusChannelsTx,
+    mut ibus_tx: IbusChannelsTx,
     ibus_rx: IbusReceiver,
 ) -> NbDaemonSender {
     let (nb_daemon_tx, nb_daemon_rx) = mpsc::channel(4);
+    ibus_tx.subscriber = Some(IbusSubscriber::new(ibus_tx.keychain.clone()));
 
     tokio::spawn(async move {
         let span = Master::debug_span("");

holo-policy/src/lib.rs

@@ -13,7 +13,7 @@ use holo_northbound::{
     NbDaemonReceiver, NbDaemonSender, NbProviderSender, ProviderBase,
     process_northbound_msg,
 };
-use holo_utils::ibus::{IbusChannelsTx, IbusReceiver};
+use holo_utils::ibus::{IbusChannelsTx, IbusReceiver, IbusSubscriber};
 use holo_utils::policy::{MatchSets, Policy};
 use tokio::sync::mpsc;
 use tracing::Instrument;
@@ -64,10 +64,11 @@ impl Master {
 
 pub fn start(
     nb_provider_tx: NbProviderSender,
-    ibus_tx: IbusChannelsTx,
+    mut ibus_tx: IbusChannelsTx,
     ibus_rx: IbusReceiver,
 ) -> NbDaemonSender {
     let (nb_daemon_tx, nb_daemon_rx) = mpsc::channel(4);
+    ibus_tx.subscriber = Some(IbusSubscriber::new(ibus_tx.policy.clone()));
 
     tokio::spawn(async move {
         let span = Master::debug_span("");

holo-protocol/src/lib.rs

@@ -17,7 +17,9 @@ use holo_northbound::{
     NbDaemonReceiver, NbDaemonSender, NbProviderSender, process_northbound_msg,
 };
 use holo_utils::bier::BierCfg;
-use holo_utils::ibus::{IbusChannelsTx, IbusMsg, IbusReceiver};
+use holo_utils::ibus::{
+    IbusChannelsTx, IbusMsg, IbusReceiver, IbusSender, IbusSubscriber,
+};
 use holo_utils::keychain::Keychains;
 use holo_utils::mpls::LabelManager;
 use holo_utils::policy::{MatchSets, Policies};
@@ -318,7 +320,7 @@ async fn run<P>(
     // Create instance Tx/Rx channels.
     let instance_channels_tx = InstanceChannelsTx::new(
         nb_tx,
-        ibus_tx,
+        ibus_tx.clone(),
         proto_input_tx,
         #[cfg(feature = "testing")]
         proto_output_tx,
@@ -353,6 +355,19 @@ async fn run<P>(
     )
     .await;
 
+    // Cancel ibus subscriptions.
+    for tx in &[
+        &ibus_tx.routing,
+        &ibus_tx.interface,
+        &ibus_tx.system,
+        &ibus_tx.keychain,
+        &ibus_tx.policy,
+    ] {
+        let _ = tx.send(IbusMsg::Disconnect {
+            subscriber: ibus_tx.subscriber.clone(),
+        });
+    }
+
     // Ensure instance is shut down before exiting.
     instance.shutdown().await;
 }
@@ -363,6 +378,7 @@ pub fn spawn_protocol_task<P>(
     name: String,
     nb_provider_tx: &NbProviderSender,
     ibus_tx: &IbusChannelsTx,
+    ibus_instance_tx: IbusSender,
     ibus_instance_rx: IbusReceiver,
     agg_channels: InstanceAggChannels<P>,
     #[cfg(feature = "testing")] test_rx: Receiver<
@@ -375,7 +391,8 @@ where
 {
     let (nb_daemon_tx, nb_daemon_rx) = mpsc::channel(4);
     let nb_provider_tx = nb_provider_tx.clone();
-    let ibus_tx = ibus_tx.clone();
+    let mut ibus_tx = ibus_tx.clone();
+    ibus_tx.subscriber = Some(IbusSubscriber::new(ibus_instance_tx));
 
     tokio::spawn(async move {
         let span = P::debug_span(&name);

holo-protocol/src/test/stub/mod.rs

@@ -262,14 +262,15 @@ where
     // Spawn protocol task.
     let (nb_provider_tx, nb_provider_rx) = mpsc::unbounded_channel();
     let (ibus_ctx, ibus_crx) = ibus::ibus_channels();
-    let (_ibus_instance_tx, ibus_instance_rx) = mpsc::unbounded_channel();
+    let (ibus_instance_tx, ibus_instance_rx) = mpsc::unbounded_channel();
     let channels = InstanceAggChannels::default();
     let instance_tx = channels.tx.clone();
     let (test_tx, test_rx) = mpsc::channel(4);
     let nb_daemon_tx = spawn_protocol_task::<P>(
         name.to_owned(),
         &nb_provider_tx,
         &ibus_ctx,
+        ibus_instance_tx,
         ibus_instance_rx,
         channels,
         test_rx,

holo-routing/src/lib.rs

@@ -20,7 +20,9 @@ use holo_northbound::{
 };
 use holo_protocol::InstanceShared;
 use holo_utils::bier::BierCfg;
-use holo_utils::ibus::{IbusChannelsTx, IbusReceiver, IbusSender};
+use holo_utils::ibus::{
+    IbusChannelsTx, IbusReceiver, IbusSender, IbusSubscriber,
+};
 use holo_utils::protocol::Protocol;
 use holo_utils::southbound::InterfaceFlags;
 use holo_utils::sr::SrCfg;
@@ -117,11 +119,12 @@ impl Master {
 
 pub fn start(
     nb_tx: NbProviderSender,
-    ibus_tx: IbusChannelsTx,
+    mut ibus_tx: IbusChannelsTx,
     ibus_rx: IbusReceiver,
     shared: InstanceShared,
 ) -> NbDaemonSender {
     let (nb_daemon_tx, nb_daemon_rx) = mpsc::channel(4);
+    ibus_tx.subscriber = Some(IbusSubscriber::new(ibus_tx.routing.clone()));
 
     tokio::spawn(async move {
         let mut master = Master {
@@ -154,6 +157,7 @@ pub fn start(
                 name,
                 &master.nb_tx,
                 &master.ibus_tx,
+                ibus_instance_tx.clone(),
                 ibus_instance_rx,
                 Default::default(),
                 shared,

holo-routing/src/northbound/configuration.rs

@@ -1129,6 +1129,7 @@ fn instance_start(master: &mut Master, protocol: Protocol, name: String) {
                 name,
                 &master.nb_tx,
                 &master.ibus_tx,
+                ibus_instance_tx.clone(),
                 ibus_instance_rx,
                 Default::default(),
                 master.shared.clone(),
@@ -1146,6 +1147,7 @@ fn instance_start(master: &mut Master, protocol: Protocol, name: String) {
                 name,
                 &master.nb_tx,
                 &master.ibus_tx,
+                ibus_instance_tx.clone(),
                 ibus_instance_rx,
                 Default::default(),
                 master.shared.clone(),
@@ -1159,6 +1161,7 @@ fn instance_start(master: &mut Master, protocol: Protocol, name: String) {
                 name,
                 &master.nb_tx,
                 &master.ibus_tx,
+                ibus_instance_tx.clone(),
                 ibus_instance_rx,
                 Default::default(),
                 master.shared.clone(),
@@ -1173,6 +1176,7 @@ fn instance_start(master: &mut Master, protocol: Protocol, name: String) {
                 name,
                 &master.nb_tx,
                 &master.ibus_tx,
+                ibus_instance_tx.clone(),
                 ibus_instance_rx,
                 Default::default(),
                 master.shared.clone(),
@@ -1187,6 +1191,7 @@ fn instance_start(master: &mut Master, protocol: Protocol, name: String) {
                 name,
                 &master.nb_tx,
                 &master.ibus_tx,
+                ibus_instance_tx.clone(),
                 ibus_instance_rx,
                 Default::default(),
                 master.shared.clone(),
@@ -1201,6 +1206,7 @@ fn instance_start(master: &mut Master, protocol: Protocol, name: String) {
                 name,
                 &master.nb_tx,
                 &master.ibus_tx,
+                ibus_instance_tx.clone(),
                 ibus_instance_rx,
                 Default::default(),
                 master.shared.clone(),
@@ -1215,6 +1221,7 @@ fn instance_start(master: &mut Master, protocol: Protocol, name: String) {
                 name,
                 &master.nb_tx,
                 &master.ibus_tx,
+                ibus_instance_tx.clone(),
                 ibus_instance_rx,
                 Default::default(),
                 master.shared.clone(),

holo-system/src/lib.rs

@@ -13,7 +13,7 @@ use holo_northbound::{
     NbDaemonReceiver, NbDaemonSender, NbProviderSender, ProviderBase,
     process_northbound_msg,
 };
-use holo_utils::ibus::{IbusChannelsTx, IbusReceiver};
+use holo_utils::ibus::{IbusChannelsTx, IbusReceiver, IbusSubscriber};
 use northbound::configuration::SystemCfg;
 use tokio::sync::mpsc;
 use tracing::Instrument;
@@ -60,10 +60,11 @@ impl Master {
 
 pub fn start(
     nb_tx: NbProviderSender,
-    ibus_tx: IbusChannelsTx,
+    mut ibus_tx: IbusChannelsTx,
     ibus_rx: IbusReceiver,
 ) -> NbDaemonSender {
     let (nb_daemon_tx, nb_daemon_rx) = mpsc::channel(4);
+    ibus_tx.subscriber = Some(IbusSubscriber::new(ibus_tx.system.clone()));
 
     tokio::spawn(async move {
         let mut master = Master {

holo-utils/src/ibus.rs

@@ -6,6 +6,7 @@
 
 use std::net::{IpAddr, Ipv4Addr};
 use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
 
 use serde::{Deserialize, Serialize};
 use tokio::sync::mpsc;
@@ -30,6 +31,7 @@ pub type IbusSender = UnboundedSender<IbusMsg>;
 // Ibus output channels.
 #[derive(Clone, Debug)]
 pub struct IbusChannelsTx {
+    pub subscriber: Option<IbusSubscriber>,
     pub routing: UnboundedSender<IbusMsg>,
     pub interface: UnboundedSender<IbusMsg>,
     pub system: UnboundedSender<IbusMsg>,
@@ -47,6 +49,15 @@ pub struct IbusChannelsRx {
     pub policy: UnboundedReceiver<IbusMsg>,
 }
 
+// Subscriber to Ibus messages.
+#[derive(Clone, Debug)]
+pub struct IbusSubscriber {
+    // Unique identifier for the subscriber.
+    pub id: usize,
+    // Channel for sending messages to the subscriber.
+    pub tx: IbusSender,
+}
+
 // Ibus message for communication among the different Holo components.
 #[derive(Clone, Debug, Deserialize, Serialize)]
 pub enum IbusMsg {
@@ -150,6 +161,23 @@ pub enum IbusMsg {
     // purged. E.g., One could ask to purge the BIRT populated by a specific
     // instance of OSPFv3 but not those populated by IS-IS.
     BierPurge,
+    // Cancel all previously requested subscriptions.
+    Disconnect {
+        #[serde(skip)]
+        subscriber: Option<IbusSubscriber>,
+    },
+}
+
+// ===== impl IbusSubscriber =====
+
+impl IbusSubscriber {
+    pub fn new(tx: IbusSender) -> Self {
+        static NEXT_ID: AtomicUsize = AtomicUsize::new(0);
+        IbusSubscriber {
+            id: NEXT_ID.fetch_add(1, Ordering::Relaxed),
+            tx,
+        }
+    }
 }
 
 // ===== global functions =====
@@ -162,6 +190,7 @@ pub fn ibus_channels() -> (IbusChannelsTx, IbusChannelsRx) {
     let (policy_tx, policy_rx) = mpsc::unbounded_channel();
 
     let tx = IbusChannelsTx {
+        subscriber: None,
         routing: routing_tx,
         interface: interface_tx,
         system: system_tx,