dynamo_runtime/component/
namespace.rs

1// SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4use anyhow::Context;
5use async_trait::async_trait;
6use futures::stream::StreamExt;
7use futures::{Stream, TryStreamExt};
8
9use super::*;
10use crate::metrics::MetricsRegistry;
11use crate::traits::events::{EventPublisher, EventSubscriber};
12
13#[async_trait]
14impl EventPublisher for Namespace {
15    fn subject(&self) -> String {
16        format!("namespace.{}", self.name)
17    }
18
19    async fn publish(
20        &self,
21        event_name: impl AsRef<str> + Send + Sync,
22        event: &(impl Serialize + Send + Sync),
23    ) -> Result<()> {
24        let bytes = serde_json::to_vec(event)?;
25        self.publish_bytes(event_name, bytes).await
26    }
27
28    async fn publish_bytes(
29        &self,
30        event_name: impl AsRef<str> + Send + Sync,
31        bytes: Vec<u8>,
32    ) -> Result<()> {
33        let subject = format!("{}.{}", self.subject(), event_name.as_ref());
34        Ok(self
35            .drt()
36            .nats_client()
37            .client()
38            .publish(subject, bytes.into())
39            .await?)
40    }
41}
42
43#[async_trait]
44impl EventSubscriber for Namespace {
45    async fn subscribe(
46        &self,
47        event_name: impl AsRef<str> + Send + Sync,
48    ) -> Result<async_nats::Subscriber> {
49        let subject = format!("{}.{}", self.subject(), event_name.as_ref());
50        Ok(self.drt().nats_client().client().subscribe(subject).await?)
51    }
52
53    async fn subscribe_with_type<T: for<'de> Deserialize<'de> + Send + 'static>(
54        &self,
55        event_name: impl AsRef<str> + Send + Sync,
56    ) -> Result<impl Stream<Item = Result<T>> + Send> {
57        let subscriber = self.subscribe(event_name).await?;
58
59        // Transform the subscriber into a stream of deserialized events
60        let stream = subscriber.map(move |msg| {
61            serde_json::from_slice::<T>(&msg.payload)
62                .with_context(|| format!("Failed to deserialize event payload: {:?}", msg.payload))
63        });
64
65        Ok(stream)
66    }
67}
68
69impl MetricsRegistry for Namespace {
70    fn basename(&self) -> String {
71        self.name.clone()
72    }
73
74    fn parent_hierarchy(&self) -> Vec<String> {
75        // Build as: [ "" (DRT), non-empty parent basenames from root -> leaf ]
76        let mut names = vec![String::new()]; // Start with empty string for DRT
77
78        // Collect parent basenames from root to leaf
79        let parent_names: Vec<String> =
80            std::iter::successors(self.parent.as_deref(), |ns| ns.parent.as_deref())
81                .map(|ns| ns.basename())
82                .filter(|name| !name.is_empty())
83                .collect();
84
85        // Append parent names in reverse order (root to leaf)
86        names.extend(parent_names.into_iter().rev());
87        names
88    }
89}
90
91#[cfg(feature = "integration")]
92#[cfg(test)]
93mod tests {
94    use super::*;
95
96    // todo - make a distributed runtime fixture
97    // todo - two options - fully mocked or integration test
98    #[tokio::test]
99    async fn test_publish() {
100        let rt = Runtime::from_current().unwrap();
101        let dtr = DistributedRuntime::from_settings(rt.clone()).await.unwrap();
102        let ns = dtr.namespace("test_namespace_publish".to_string()).unwrap();
103        ns.publish("test_event", &"test".to_string()).await.unwrap();
104        rt.shutdown();
105    }
106
107    #[tokio::test]
108    async fn test_subscribe() {
109        let rt = Runtime::from_current().unwrap();
110        let dtr = DistributedRuntime::from_settings(rt.clone()).await.unwrap();
111        let ns = dtr
112            .namespace("test_namespace_subscribe".to_string())
113            .unwrap();
114
115        // Create a subscriber
116        let mut subscriber = ns.subscribe("test_event").await.unwrap();
117
118        // Publish a message
119        ns.publish("test_event", &"test_message".to_string())
120            .await
121            .unwrap();
122
123        // Receive the message
124        if let Some(msg) = subscriber.next().await {
125            let received = String::from_utf8(msg.payload.to_vec()).unwrap();
126            assert_eq!(received, "\"test_message\"");
127        }
128
129        rt.shutdown();
130    }
131}
dynamo_runtime/component/namespace.rs

dynamo_runtime/component/
namespace.rs
