feat(yang-push): separate transport session info from subscription identity

assets/yang/ietf-interfaces/subscriptions-info.json

@@ -1,7 +1,6 @@
 [
   {
-    "collector": "127.0.0.1:10000",
-    "peer": "0.0.0.0:830",
+    "peer_ip": "0.0.0.0",
     "id": 1,
     "target": {
       "ietf-yang-push:datastore": "ietf-datastores:operational",

assets/yang/ietf-telemetry-message/subscriptions-info.json

@@ -1,6 +1,6 @@
 [
   {
-    "peer": "0.0.0.0:830",
+    "peer_ip": "0.0.0.0",
     "id": 1,
     "content_id": "ietf-telemetry-message",
     "target": {

crates/collector/examples/kafka-yang-consumer.rs

@@ -46,7 +46,7 @@ use schema_registry_client::rest::schema_registry_client::{Client, SchemaRegistr
 use serde_json::json;
 use shadow_rs::shadow;
 use std::collections::{HashMap, HashSet};
-use std::net::{IpAddr, SocketAddr};
+use std::net::IpAddr;
 use tokio::signal;
 use tracing::{debug, error, info, trace, warn};
 use yang5::context::Context;
@@ -845,12 +845,8 @@ async fn main() -> Result<()> {
     let sr_client = SchemaRegistryClient::new(sr_config);
 
     // Create placeholder subscription info (reused for all schemas)
-    let subscription_info = SubscriptionInfo::new_empty(
-        SocketAddr::new(IpAddr::V4(std::net::Ipv4Addr::new(0, 0, 0, 0)), 0),
-        None,
-        SocketAddr::new(IpAddr::V4(std::net::Ipv4Addr::new(0, 0, 0, 0)), 0),
-        0,
-    );
+    let subscription_info =
+        SubscriptionInfo::new_empty(IpAddr::V4(std::net::Ipv4Addr::new(0, 0, 0, 0)), 0);
 
     // Create YANG context cache
     let mut yang_ctx_cache = YangContextCache::new();

crates/collector/src/lib.rs

@@ -1,3 +1,4 @@
+// Copyright (C) 2026-present The NetCalyx Authors.
 // Copyright (C) 2024-present The NetGauze Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
@@ -24,7 +25,7 @@ use crate::publishers::http::{HttpPublisherActorHandle, Message};
 use crate::publishers::kafka_avro::KafkaAvroPublisherActorHandle;
 use crate::publishers::kafka_json::KafkaJsonPublisherActorHandle;
 use crate::publishers::kafka_yang::KafkaYangPublisherActorHandle;
-use crate::yang_push::enrichment::YangPushEnrichmentActorHandle;
+use crate::yang_push::enrichment::{EnrichedNotification, YangPushEnrichmentActorHandle};
 
 use futures_util::StreamExt;
 use futures_util::stream::FuturesUnordered;
@@ -36,11 +37,9 @@ use netcalyx_flow_service::flow_supervisor::FlowCollectorsSupervisorActorHandle;
 use netcalyx_udp_notif_pkt::raw::MediaType;
 use netcalyx_udp_notif_service::UdpNotifRequest;
 use netcalyx_udp_notif_service::supervisor::UdpNotifSupervisorHandle;
-use netcalyx_yang_push::ContentId;
 use netcalyx_yang_push::cache::actor::CacheActorHandle;
 use netcalyx_yang_push::cache::fetcher::{NetconfYangLibraryFetcher, RetryConfig};
-use netcalyx_yang_push::cache::storage::SubscriptionInfo;
-use netcalyx_yang_push::model::telemetry::{Manifest, TelemetryMessageWrapper};
+use netcalyx_yang_push::model::telemetry::Manifest;
 use netcalyx_yang_push::validation::ValidationActorHandle;
 use shadow_rs::shadow;
 use std::net::IpAddr;
@@ -926,10 +925,10 @@ fn serialize_udp_notif(
 }
 
 fn serialize_telemetry_json(
-    input: (Option<ContentId>, SubscriptionInfo, TelemetryMessageWrapper),
+    input: EnrichedNotification,
     _writer_id: String,
 ) -> Result<(Option<serde_json::Value>, serde_json::Value), UdpNotifSerializationError> {
-    let tmw = input.2;
+    let tmw = input.message;
     let ip = tmw.message().telemetry_message_metadata().export_address();
     let value = serde_json::to_value(tmw)?;
     let key = serde_json::Value::String(ip.to_string());
@@ -1054,6 +1053,7 @@ mod tests {
     use super::*;
     use bytes::Bytes;
     use netcalyx_udp_notif_pkt::raw::UdpNotifPacket;
+    use netcalyx_udp_notif_service::SessionInfo;
     use std::collections::HashMap;
     use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 
@@ -1070,7 +1070,10 @@ mod tests {
             Bytes::from(&[0xffu8, 0xffu8][..]),
         );
 
-        let request = Arc::new(UdpNotifRequest::new(collector, None, peer, pkt));
+        let request = Arc::new(UdpNotifRequest::new(
+            SessionInfo::new(collector, None, peer),
+            pkt,
+        ));
         let serialized = serialize_udp_notif(request.clone(), writer_id.clone());
         assert!(matches!(
             serialized,
@@ -1112,12 +1115,13 @@ mod tests {
             }
         );
         let request_invalid = Arc::new(UdpNotifRequest::new(
-            collector,
-            None,
-            peer,
+            SessionInfo::new(collector, None, peer),
             pkt_invalid_json,
         ));
-        let request_good = Arc::new(UdpNotifRequest::new(collector, None, peer, pkt));
+        let request_good = Arc::new(UdpNotifRequest::new(
+            SessionInfo::new(collector, None, peer),
+            pkt,
+        ));
         let result_invalid = serialize_udp_notif(request_invalid, writer_id.clone());
         let serialized =
             serialize_udp_notif(request_good, writer_id.clone()).expect("failed to serialize json");
@@ -1168,12 +1172,13 @@ mod tests {
         );
 
         let request_invalid = Arc::new(UdpNotifRequest::new(
-            collector,
-            None,
-            peer,
+            SessionInfo::new(collector, None, peer),
             pkt_invalid_utf8,
         ));
-        let request_good = Arc::new(UdpNotifRequest::new(collector, None, peer, pkt));
+        let request_good = Arc::new(UdpNotifRequest::new(
+            SessionInfo::new(collector, None, peer),
+            pkt,
+        ));
         let result_invalid = serialize_udp_notif(request_invalid, writer_id.clone());
         let serialized =
             serialize_udp_notif(request_good, writer_id.clone()).expect("failed to serialize json");
@@ -1226,8 +1231,14 @@ mod tests {
             }
         );
 
-        let request_invalid = Arc::new(UdpNotifRequest::new(collector, None, peer, pkt_invalid));
-        let request_good = Arc::new(UdpNotifRequest::new(collector, None, peer, pkt));
+        let request_invalid = Arc::new(UdpNotifRequest::new(
+            SessionInfo::new(collector, None, peer),
+            pkt_invalid,
+        ));
+        let request_good = Arc::new(UdpNotifRequest::new(
+            SessionInfo::new(collector, None, peer),
+            pkt,
+        ));
         let result_invalid = serialize_udp_notif(request_invalid, writer_id.clone());
         let serialized =
             serialize_udp_notif(request_good, writer_id.clone()).expect("failed to serialize json");

crates/collector/src/publishers/kafka_yang.rs

@@ -1,3 +1,4 @@
+// Copyright (C) 2026-present The NetCalyx Authors.
 // Copyright (C) 2025-present The NetGauze Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
@@ -430,7 +431,7 @@ where
             return if let Some(default_schema_id) = self.default_schema_id {
                 if let Some(subscription_info) = subscription_info {
                     warn!(
-                        peer=%subscription_info.peer(),
+                        peer_ip=%subscription_info.peer_ip(),
                         subscription_id=subscription_info.id(),
                         router_content_id=subscription_info.content_id(),
                         target=%subscription_info.target(),
@@ -449,7 +450,7 @@ where
             } else {
                 if let Some(subscription_info) = subscription_info {
                     warn!(
-                        peer=%subscription_info.peer(),
+                        peer_ip=%subscription_info.peer_ip(),
                         subscription_id=subscription_info.id(),
                         router_content_id=subscription_info.content_id(),
                         target=%subscription_info.target(),
@@ -470,7 +471,7 @@ where
         if let Some(&schema_id) = self.schema_id_cache.get(id) {
             if let Some(subscription_info) = subscription_info {
                 trace!(
-                    peer=%subscription_info.peer(),
+                    peer_ip=%subscription_info.peer_ip(),
                     subscription_id=subscription_info.id(),
                     router_content_id=subscription_info.content_id(),
                     target=%subscription_info.target(),
@@ -495,10 +496,10 @@ where
 
         if let Err(err) = self
             .cache_req_tx
-            .send(CacheLookupCommand::LookupByContentIdOneShot(
-                id.to_string(),
-                response_tx,
-            ))
+            .send(CacheLookupCommand::LookupByContentIdOneShot {
+                content_id: id.to_string(),
+                tx: response_tx,
+            })
             .await
         {
             warn!("Failed to request schema for content_id: {}", id);

crates/collector/src/yang_push/config.rs

@@ -1,3 +1,4 @@
+// Copyright (C) 2026-present The NetCalyx Authors.
 // Copyright (C) 2025-present The NetGauze Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
@@ -21,9 +22,9 @@
 //! JSON format.
 
 use crate::publishers::kafka_yang::YangConverter;
+use crate::yang_push::enrichment::EnrichedNotification;
 use netcalyx_yang_push::ContentId;
 use netcalyx_yang_push::cache::storage::{SubscriptionInfo, YangLibraryReference};
-use netcalyx_yang_push::model::telemetry::TelemetryMessageWrapper;
 use serde::{Deserialize, Serialize};
 
 #[derive(Debug, strum_macros::Display)]
@@ -81,12 +82,7 @@ impl TelemetryYangConverter {
     }
 }
 
-impl
-    YangConverter<
-        (Option<ContentId>, SubscriptionInfo, TelemetryMessageWrapper),
-        TelemetryYangConverterError,
-    > for TelemetryYangConverter
-{
+impl YangConverter<EnrichedNotification, TelemetryYangConverterError> for TelemetryYangConverter {
     fn subject_prefix(&self) -> Option<&str> {
         self.subject_prefix.as_deref()
     }
@@ -103,38 +99,27 @@ impl
         self.extension_yang_lib_ref.as_ref()
     }
 
-    fn content_id(
-        &self,
-        input: &(Option<ContentId>, SubscriptionInfo, TelemetryMessageWrapper),
-    ) -> Option<ContentId> {
-        input.0.clone()
+    fn content_id(&self, input: &EnrichedNotification) -> Option<ContentId> {
+        input.cached_content_id.clone()
     }
 
-    fn get_key(
-        &self,
-        input: &(Option<ContentId>, SubscriptionInfo, TelemetryMessageWrapper),
-    ) -> Option<serde_json::Value> {
-        let (_, subscription_info, _) = input;
-        let ip = subscription_info.peer().ip();
+    fn get_key(&self, input: &EnrichedNotification) -> Option<serde_json::Value> {
+        let ip = input.subscription_info.peer_ip();
         Some(serde_json::Value::String(ip.to_string()))
     }
 
     fn serialize_json(
         &self,
-        input: (Option<ContentId>, SubscriptionInfo, TelemetryMessageWrapper),
+        input: EnrichedNotification,
     ) -> Result<Vec<u8>, TelemetryYangConverterError> {
-        let telemetry_message_wrapper = input.2;
-        serde_json::to_vec(&telemetry_message_wrapper).map_err(Into::into)
+        serde_json::to_vec(&input.message).map_err(Into::into)
     }
 
-    fn subscription_info(
-        &self,
-        input: &(Option<ContentId>, SubscriptionInfo, TelemetryMessageWrapper),
-    ) -> Option<SubscriptionInfo> {
-        if input.1.is_empty() {
+    fn subscription_info(&self, input: &EnrichedNotification) -> Option<SubscriptionInfo> {
+        if input.subscription_info.is_empty() {
             None
         } else {
-            Some(input.1.clone())
+            Some(input.subscription_info.clone())
         }
     }
 }
@@ -145,14 +130,13 @@ mod tests {
     use chrono::TimeZone;
     use netcalyx_netconf_proto::yang_push::identities::{Encoding, Transport};
     use netcalyx_netconf_proto::yang_push::subscription::YangPushModuleVersion;
+    use netcalyx_udp_notif_service::SessionInfo;
     use netcalyx_yang_push::model::telemetry::*;
     use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
 
     fn create_test_subscription_info(ip: IpAddr) -> SubscriptionInfo {
         SubscriptionInfo::new(
-            SocketAddr::from(([127, 0, 0, 1], 10000)),
-            None,
-            SocketAddr::new(ip, 8080),
+            ip,
             1,
             netcalyx_udp_notif_pkt::notification::Target::new_datastore(
                 "ietf-datastores:operational".to_string(),
@@ -223,7 +207,16 @@ mod tests {
         let sub_info = create_test_subscription_info(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)));
         let msg = create_test_telemetry_message_wrapper();
 
-        let input = (content_id.clone(), sub_info, msg);
+        let input = EnrichedNotification {
+            cached_content_id: content_id.clone(),
+            subscription_info: sub_info,
+            session: SessionInfo::new(
+                SocketAddr::from(([127, 0, 0, 1], 0)),
+                None,
+                SocketAddr::from(([127, 0, 0, 1], 0)),
+            ),
+            message: msg,
+        };
         assert_eq!(converter.content_id(&input), content_id);
     }
 
@@ -234,7 +227,16 @@ mod tests {
         let sub_info = create_test_subscription_info(ip);
         let msg = create_test_telemetry_message_wrapper();
 
-        let input = (None, sub_info, msg);
+        let input = EnrichedNotification {
+            cached_content_id: None,
+            subscription_info: sub_info,
+            session: SessionInfo::new(
+                SocketAddr::from(([127, 0, 0, 1], 0)),
+                None,
+                SocketAddr::from(([127, 0, 0, 1], 0)),
+            ),
+            message: msg,
+        };
         let key = converter.get_key(&input).unwrap();
         assert_eq!(key, serde_json::Value::String("192.168.1.1".to_string()));
     }
@@ -246,7 +248,16 @@ mod tests {
         let sub_info = create_test_subscription_info(ip);
         let msg = create_test_telemetry_message_wrapper();
 
-        let input = (None, sub_info, msg);
+        let input = EnrichedNotification {
+            cached_content_id: None,
+            subscription_info: sub_info,
+            session: SessionInfo::new(
+                SocketAddr::from(([127, 0, 0, 1], 0)),
+                None,
+                SocketAddr::from(([127, 0, 0, 1], 0)),
+            ),
+            message: msg,
+        };
         let key = converter.get_key(&input).unwrap();
         assert_eq!(key, serde_json::Value::String("2001:db8::1".to_string()));
     }
@@ -259,7 +270,16 @@ mod tests {
         let expected = serde_json::to_value(&msg).unwrap();
 
         // Call serialize_json to serialize into bytes
-        let input = (None, sub_info, msg);
+        let input = EnrichedNotification {
+            cached_content_id: None,
+            subscription_info: sub_info,
+            session: SessionInfo::new(
+                SocketAddr::from(([127, 0, 0, 1], 0)),
+                None,
+                SocketAddr::from(([127, 0, 0, 1], 0)),
+            ),
+            message: msg,
+        };
         let result = converter.serialize_json(input);
         assert!(result.is_ok());