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//
4// Licensed under the Apache License, Version 2.0 (the "License");
5// you may not use this file except in compliance with the License.
6// You may obtain a copy of the License at
7//
8// http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15
16use anyhow::Context;
17use async_trait::async_trait;
18use futures::stream::StreamExt;
19use futures::{Stream, TryStreamExt};
20
21use super::*;
22use crate::metrics::MetricsRegistry;
23use crate::traits::events::{EventPublisher, EventSubscriber};
24
25#[async_trait]
26impl EventPublisher for Namespace {
27    fn subject(&self) -> String {
28        format!("namespace.{}", self.name)
29    }
30
31    async fn publish(
32        &self,
33        event_name: impl AsRef<str> + Send + Sync,
34        event: &(impl Serialize + Send + Sync),
35    ) -> Result<()> {
36        let bytes = serde_json::to_vec(event)?;
37        self.publish_bytes(event_name, bytes).await
38    }
39
40    async fn publish_bytes(
41        &self,
42        event_name: impl AsRef<str> + Send + Sync,
43        bytes: Vec<u8>,
44    ) -> Result<()> {
45        let subject = format!("{}.{}", self.subject(), event_name.as_ref());
46        Ok(self
47            .drt()
48            .nats_client()
49            .client()
50            .publish(subject, bytes.into())
51            .await?)
52    }
53}
54
55#[async_trait]
56impl EventSubscriber for Namespace {
57    async fn subscribe(
58        &self,
59        event_name: impl AsRef<str> + Send + Sync,
60    ) -> Result<async_nats::Subscriber> {
61        let subject = format!("{}.{}", self.subject(), event_name.as_ref());
62        Ok(self.drt().nats_client().client().subscribe(subject).await?)
63    }
64
65    async fn subscribe_with_type<T: for<'de> Deserialize<'de> + Send + 'static>(
66        &self,
67        event_name: impl AsRef<str> + Send + Sync,
68    ) -> Result<impl Stream<Item = Result<T>> + Send> {
69        let subscriber = self.subscribe(event_name).await?;
70
71        // Transform the subscriber into a stream of deserialized events
72        let stream = subscriber.map(move |msg| {
73            serde_json::from_slice::<T>(&msg.payload)
74                .with_context(|| format!("Failed to deserialize event payload: {:?}", msg.payload))
75        });
76
77        Ok(stream)
78    }
79}
80
81impl MetricsRegistry for Namespace {
82    fn basename(&self) -> String {
83        self.name.clone()
84    }
85
86    fn parent_hierarchy(&self) -> Vec<String> {
87        // Build as: [ "" (DRT), non-empty parent basenames from root -> leaf ]
88        let mut names = vec![String::new()]; // Start with empty string for DRT
89
90        // Collect parent basenames from root to leaf
91        let parent_names: Vec<String> =
92            std::iter::successors(self.parent.as_deref(), |ns| ns.parent.as_deref())
93                .map(|ns| ns.basename())
94                .filter(|name| !name.is_empty())
95                .collect();
96
97        // Append parent names in reverse order (root to leaf)
98        names.extend(parent_names.into_iter().rev());
99        names
100    }
101}
102
103#[cfg(feature = "integration")]
104#[cfg(test)]
105mod tests {
106    use super::*;
107
108    // todo - make a distributed runtime fixture
109    // todo - two options - fully mocked or integration test
110    #[tokio::test]
111    async fn test_publish() {
112        let rt = Runtime::from_current().unwrap();
113        let dtr = DistributedRuntime::from_settings(rt.clone()).await.unwrap();
114        let ns = dtr.namespace("test_namespace_publish".to_string()).unwrap();
115        ns.publish("test_event", &"test".to_string()).await.unwrap();
116        rt.shutdown();
117    }
118
119    #[tokio::test]
120    async fn test_subscribe() {
121        let rt = Runtime::from_current().unwrap();
122        let dtr = DistributedRuntime::from_settings(rt.clone()).await.unwrap();
123        let ns = dtr
124            .namespace("test_namespace_subscribe".to_string())
125            .unwrap();
126
127        // Create a subscriber
128        let mut subscriber = ns.subscribe("test_event").await.unwrap();
129
130        // Publish a message
131        ns.publish("test_event", &"test_message".to_string())
132            .await
133            .unwrap();
134
135        // Receive the message
136        if let Some(msg) = subscriber.next().await {
137            let received = String::from_utf8(msg.payload.to_vec()).unwrap();
138            assert_eq!(received, "\"test_message\"");
139        }
140
141        rt.shutdown();
142    }
143}
dynamo_runtime/component/namespace.rs

dynamo_runtime/component/
namespace.rs
