up_rust/communication/
simple_notifier.rs

1/********************************************************************************
2 * Copyright (c) 2024 Contributors to the Eclipse Foundation
3 *
4 * See the NOTICE file(s) distributed with this work for additional
5 * information regarding copyright ownership.
6 *
7 * This program and the accompanying materials are made available under the
8 * terms of the Apache License Version 2.0 which is available at
9 * https://www.apache.org/licenses/LICENSE-2.0
10 *
11 * SPDX-License-Identifier: Apache-2.0
12 ********************************************************************************/
13
14// [impl->dsn~communication-layer-impl-default~1]
15
16use std::sync::Arc;
17
18use async_trait::async_trait;
19
20use crate::{
21    communication::{
22        apply_common_options, build_message, CallOptions, NotificationError, Notifier,
23        RegistrationError, UPayload,
24    },
25    LocalUriProvider, UListener, UMessageBuilder, UTransport, UUri,
26};
27
28/// A [`Notifier`] that uses the uProtocol Transport Layer API to send and receive
29/// notifications to/from (other) uEntities.
30pub struct SimpleNotifier<T, P> {
31    transport: Arc<T>,
32    uri_provider: Arc<P>,
33}
34
35impl<T, P> core::fmt::Debug for SimpleNotifier<T, P> {
36    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
37        f.debug_struct("SimpleNotifier").finish_non_exhaustive()
38    }
39}
40
41impl<T: UTransport, P: LocalUriProvider> SimpleNotifier<T, P> {
42    /// Creates a new Notifier for a given transport.
43    ///
44    /// # Arguments
45    ///
46    /// * `transport` - The uProtocol Transport Layer implementation to use for sending and receiving notification messages.
47    /// * `uri_provider` - The helper for creating URIs that represent local resources.
48    pub fn new(transport: Arc<T>, uri_provider: Arc<P>) -> Self {
49        SimpleNotifier {
50            transport,
51            uri_provider,
52        }
53    }
54}
55
56#[async_trait]
57impl<T: UTransport, P: LocalUriProvider> Notifier for SimpleNotifier<T, P> {
58    async fn notify(
59        &self,
60        resource_id: u16,
61        destination: &UUri,
62        call_options: CallOptions,
63        payload: Option<UPayload>,
64    ) -> Result<(), NotificationError> {
65        let mut builder = UMessageBuilder::notification(
66            self.uri_provider.get_resource_uri(resource_id),
67            destination.to_owned(),
68        );
69        apply_common_options(call_options, &mut builder);
70        let msg = build_message(&mut builder, payload)
71            .map_err(|e| NotificationError::InvalidArgument(e.to_string()))?;
72        self.transport
73            .send(msg)
74            .await
75            .map_err(Box::from)
76            .map_err(NotificationError::NotifyError)
77    }
78
79    async fn start_listening(
80        &self,
81        topic: &UUri,
82        listener: Arc<dyn UListener>,
83    ) -> Result<(), RegistrationError> {
84        topic
85            .verify_no_wildcards()
86            .map_err(|e| RegistrationError::InvalidFilter(e.to_string()))?;
87        self.transport
88            .register_listener(topic, Some(&self.uri_provider.get_source_uri()), listener)
89            .await
90            .map_err(RegistrationError::from)
91    }
92
93    async fn stop_listening(
94        &self,
95        topic: &UUri,
96        listener: Arc<dyn UListener>,
97    ) -> Result<(), RegistrationError> {
98        topic
99            .verify_no_wildcards()
100            .map_err(|e| RegistrationError::InvalidFilter(e.to_string()))?;
101        self.transport
102            .unregister_listener(topic, Some(&self.uri_provider.get_source_uri()), listener)
103            .await
104            .map_err(RegistrationError::from)
105    }
106}
107
108#[cfg(test)]
109mod tests {
110
111    // [utest->dsn~communication-layer-impl-default~1]
112
113    use super::*;
114
115    use crate::{
116        utransport::{MockTransport, MockUListener},
117        StaticUriProvider, UCode, UPriority, UStatus, UUri, UUID,
118    };
119
120    fn new_uri_provider() -> Arc<StaticUriProvider> {
121        Arc::new(StaticUriProvider::new("", 0x0005, 0x02).expect("failed to create URI Provider"))
122    }
123
124    #[tokio::test]
125    async fn test_start_stop_listening_rejects_wildcard_topic() {
126        let mut transport = MockTransport::new();
127        transport.expect_do_register_listener().never();
128        let uri_provider = new_uri_provider();
129        let notifier = SimpleNotifier::new(Arc::new(transport), uri_provider);
130
131        let invalid_topic = UUri::try_from("up://my-vin/A15B/1/FFFF").unwrap();
132        let mut listener = MockUListener::new();
133        listener.expect_on_receive().never();
134        let wrapped_listener = Arc::new(listener);
135
136        let result = notifier
137            .start_listening(&invalid_topic, wrapped_listener.clone())
138            .await;
139        assert!(result.is_err_and(|e| matches!(e, RegistrationError::InvalidFilter(_))));
140
141        let result = notifier
142            .stop_listening(&invalid_topic, wrapped_listener)
143            .await;
144        assert!(result.is_err_and(|e| matches!(e, RegistrationError::InvalidFilter(_))));
145    }
146
147    #[tokio::test]
148    async fn test_start_listening_succeeds() {
149        let uri_provider = new_uri_provider();
150        let topic = UUri::try_from("up://my-vin/A15B/1/B10F").unwrap();
151        let expected_source_filter = topic.clone();
152        let expected_sink_filter = uri_provider.get_source_uri();
153        let mut transport = MockTransport::new();
154        transport
155            .expect_do_register_listener()
156            .once()
157            .withf(move |source_filter, sink_filter, _listener| {
158                source_filter == &expected_source_filter
159                    && *sink_filter == Some(&expected_sink_filter)
160            })
161            .return_const(Ok(()));
162        let notifier = SimpleNotifier::new(Arc::new(transport), uri_provider);
163
164        let mut listener = MockUListener::new();
165        listener.expect_on_receive().never();
166        let result = notifier.start_listening(&topic, Arc::new(listener)).await;
167        assert!(result.is_ok());
168    }
169
170    #[tokio::test]
171    async fn test_stop_listening_succeeds() {
172        let uri_provider = new_uri_provider();
173        let topic = UUri::try_from("up://my-vin/A15B/1/B10F").unwrap();
174        let expected_source_filter = topic.clone();
175        let expected_sink_filter = uri_provider.get_source_uri();
176        let mut transport = MockTransport::new();
177        transport
178            .expect_do_unregister_listener()
179            .once()
180            .withf(move |source_filter, sink_filter, _listener| {
181                source_filter == &expected_source_filter
182                    && *sink_filter == Some(&expected_sink_filter)
183            })
184            .return_const(Ok(()));
185        let notifier = SimpleNotifier::new(Arc::new(transport), uri_provider);
186
187        let mut listener = MockUListener::new();
188        listener.expect_on_receive().never();
189        let result = notifier.stop_listening(&topic, Arc::new(listener)).await;
190        assert!(result.is_ok());
191    }
192
193    #[tokio::test]
194    async fn test_publish_succeeds() {
195        let message_id = UUID::build();
196        let value = b"Hello";
197
198        let uri_provider = new_uri_provider();
199        let destination = UUri::try_from("up://other-vin/A15B/1/0").unwrap();
200        let expected_message_id = message_id.clone();
201        let expected_sink = destination.clone();
202        let expected_source = uri_provider.get_resource_uri(0xB10F);
203        let mut transport = MockTransport::new();
204        transport
205            .expect_do_send()
206            .once()
207            .withf(move |message| {
208                message.is_notification()
209                    && message.id() == &expected_message_id
210                    && message.source() == &expected_source
211                    && message.sink_unchecked() == &expected_sink
212                    && message.ttl_unchecked() == 10_000
213                    && message.priority_unchecked() == UPriority::CS2
214                    && message.payload() == Some(value.as_slice().into())
215            })
216            .return_const(Ok(()));
217        let notifier = SimpleNotifier::new(Arc::new(transport), uri_provider);
218
219        let payload = UPayload::new(value.as_slice(), crate::UPayloadFormat::Raw);
220        let options =
221            CallOptions::for_notification(Some(10_000), Some(message_id), Some(UPriority::CS2));
222        let result = notifier
223            .notify(0xB10F, &destination, options, Some(payload))
224            .await;
225        assert!(result.is_ok());
226    }
227
228    #[tokio::test]
229    async fn test_publish_fails_for_transport_error() {
230        let uri_provider = new_uri_provider();
231        let destination = UUri::try_from("up://other-vin/A15B/1/0").unwrap();
232        let mut transport = MockTransport::new();
233        transport
234            .expect_do_send()
235            .once()
236            .return_const(Err(UStatus::fail_with_code(
237                crate::UCode::Unavailable,
238                "connection lost",
239            )));
240        let notifier = SimpleNotifier::new(Arc::new(transport), uri_provider);
241
242        let options = CallOptions::for_notification(None, None, None);
243        let result = notifier.notify(0xB10F, &destination, options, None).await;
244        assert!(result.is_err_and(|e| match e {
245            NotificationError::NotifyError(status) => status.code() == UCode::Unavailable,
246            _ => false,
247        }));
248    }
249
250    #[tokio::test]
251    async fn test_publish_fails_for_invalid_destination() {
252        let uri_provider = new_uri_provider();
253        // destination has resource ID != 0
254        let destination = UUri::try_from("up://other-vin/A15B/1/10").unwrap();
255        let mut transport = MockTransport::new();
256        transport.expect_do_send().never();
257        let notifier = SimpleNotifier::new(Arc::new(transport), uri_provider);
258
259        let options = CallOptions::for_notification(None, None, None);
260        let result = notifier.notify(0xB10F, &destination, options, None).await;
261        assert!(result.is_err_and(|e| matches!(e, NotificationError::InvalidArgument(_))));
262    }
263
264    #[tokio::test]
265    async fn test_publish_fails_for_invalid_resource_id() {
266        let uri_provider = new_uri_provider();
267        let destination = UUri::try_from("up://other-vin/A15B/1/0").unwrap();
268        let mut transport = MockTransport::new();
269        transport.expect_do_send().never();
270        let notifier = SimpleNotifier::new(Arc::new(transport), uri_provider);
271
272        let options = CallOptions::for_notification(None, None, None);
273        // resource ID of origin address must not be 0
274        let result = notifier.notify(0x0000, &destination, options, None).await;
275        assert!(result.is_err_and(|e| matches!(e, NotificationError::InvalidArgument(_))));
276    }
277}
up_rust/communication/simple_notifier.rs

up_rust/communication/
simple_notifier.rs
