up_rust/communication/
in_memory_subscriber.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::{
17    collections::{hash_map::Entry, HashMap},
18    ops::Deref,
19    sync::{Arc, RwLock},
20};
21
22use async_trait::async_trait;
23use tracing::{debug, info};
24
25use crate::{
26    communication::{
27        Notifier, RegistrationError, Subscriber, SubscriptionChangeHandler, SubscriptionStatus,
28    },
29    core::usubscription::{self, SubscriptionInfo, USubscription},
30    UListener, UMessage, UStatus, UTransport, UUri,
31};
32
33#[derive(Clone)]
34struct ComparableSubscriptionChangeHandler {
35    inner: Arc<dyn SubscriptionChangeHandler>,
36}
37
38impl ComparableSubscriptionChangeHandler {
39    fn new(handler: Arc<dyn SubscriptionChangeHandler>) -> Self {
40        ComparableSubscriptionChangeHandler {
41            inner: handler.clone(),
42        }
43    }
44}
45
46impl Deref for ComparableSubscriptionChangeHandler {
47    type Target = dyn SubscriptionChangeHandler;
48
49    fn deref(&self) -> &Self::Target {
50        &*self.inner
51    }
52}
53
54impl PartialEq for ComparableSubscriptionChangeHandler {
55    /// Compares this handler to another handler.
56    ///
57    /// # Returns
58    ///
59    /// `true` if the pointer to the handler held by `self` is equal to the pointer held by `other`.
60    /// This is consistent with the implementation of [`ComparableSubscriptionChangeHandler::hash`].
61    fn eq(&self, other: &Self) -> bool {
62        Arc::ptr_eq(&self.inner, &other.inner)
63    }
64}
65
66impl Eq for ComparableSubscriptionChangeHandler {}
67
68#[derive(Default)]
69struct SubscriptionChangeListener {
70    subscription_change_handlers: RwLock<HashMap<UUri, ComparableSubscriptionChangeHandler>>,
71}
72
73impl SubscriptionChangeListener {
74    /// Adds a handler for a given topic.
75    ///
76    /// # Errors
77    ///
78    /// Returns a [`RegistrationError::AlreadyExists`] if another handler has already been registered for
79    /// the given topic. Returns a [`RegistrationError::Unknown`] if the internal state could not be accessed,
80    fn add_handler(
81        &self,
82        topic: UUri,
83        subscription_change_handler: Arc<dyn SubscriptionChangeHandler>,
84    ) -> Result<(), RegistrationError> {
85        let Ok(mut handlers) = self.subscription_change_handlers.write() else {
86            return Err(RegistrationError::Unknown(Box::from(
87                UStatus::fail_with_code(
88                    crate::UCode::Internal,
89                    "failed to acquire write lock for handler map",
90                ),
91            )));
92        };
93        let handler_to_add = ComparableSubscriptionChangeHandler::new(subscription_change_handler);
94        match handlers.entry(topic) {
95            Entry::Vacant(entry) => {
96                entry.insert(handler_to_add);
97                Ok(())
98            }
99            Entry::Occupied(entry) => {
100                if entry.get() == &handler_to_add {
101                    Ok(())
102                } else {
103                    Err(RegistrationError::AlreadyExists)
104                }
105            }
106        }
107    }
108
109    /// Removes the handler for a given topic.
110    ///
111    /// This function also succeeds if no handler is registered for the topic.
112    ///
113    /// # Errors
114    ///
115    /// Returns a [`RegistrationError::Unknown`] if the internal state could not be accessed,
116    fn remove_handler(&self, topic: &UUri) -> Result<(), RegistrationError> {
117        self.subscription_change_handlers
118            .write()
119            .map_err(|_e| {
120                RegistrationError::Unknown(Box::from(UStatus::fail_with_code(
121                    crate::UCode::Internal,
122                    "failed to acquire write lock for handler map",
123                )))
124            })
125            .map(|mut handlers| {
126                handlers.remove(topic);
127            })
128    }
129
130    /// Removes all handlers for all topic.
131    ///
132    /// # Errors
133    ///
134    /// Returns a [`RegistrationError::Unknown`] if the internal state could not be accessed,
135    fn clear(&self) -> Result<(), RegistrationError> {
136        self.subscription_change_handlers
137            .write()
138            .map_err(|_e| {
139                RegistrationError::Unknown(Box::from(UStatus::fail_with_code(
140                    crate::UCode::Internal,
141                    "failed to acquire write lock for handler map",
142                )))
143            })
144            .map(|mut handlers| {
145                handlers.clear();
146            })
147    }
148
149    #[cfg(test)]
150    fn has_handler(&self, topic: &UUri) -> bool {
151        self.subscription_change_handlers
152            .read()
153            .is_ok_and(|handlers| handlers.contains_key(topic))
154    }
155}
156
157#[async_trait]
158impl UListener for SubscriptionChangeListener {
159    async fn on_receive(&self, msg: UMessage) {
160        if !msg.is_notification() {
161            return;
162        }
163        let Ok(subscription_update) =
164            msg.extract_protobuf::<crate::up_core_api::usubscription::Update>()
165        else {
166            debug!("ignoring notification that does not contain subscription update");
167            return;
168        };
169        let Ok(subscription_info) = SubscriptionInfo::try_from(&subscription_update) else {
170            debug!("ignoring notification that does not contain valid subscription update");
171            return;
172        };
173        let topic = subscription_info.topic().to_owned();
174
175        let Ok(handlers) = self.subscription_change_handlers.read() else {
176            return;
177        };
178        if let Some(handler) = handlers.get(&topic) {
179            handler.on_subscription_change(topic, subscription_info.status().to_owned());
180        }
181    }
182}
183
184/// A [`Subscriber`] which keeps all information about registered subscription change handlers in memory.
185///
186/// The subscriber requires a (client) implementation of [`USubscription`] in order to inform the local
187/// USubscription service about newly subscribed and unsubscribed topics. It also needs a [`Notifier`]
188/// for receiving notifications about subscription status updates from the local USubscription service.
189/// Finally, it needs a [`UTransport`] for receiving events that have been published to subscribed topics.
190///
191/// During [startup](`Self::for_clients`) the subscriber uses the Notifier to register a generic [`UListener`]
192/// for receiving notifications from the USubscription service. The listener maintains an in-memory mapping
193/// of subscribed topics to corresponding subscription change handlers.
194///
195/// When a client [`subscribes to a topic`](Self::subscribe), the local USubscription service is informed
196/// about the new subscription and a (client provided) subscription change handler is registered with the
197/// listener. When a subscription change notification arrives from the USubscription service, the corresponding
198/// handler is being looked up and invoked.
199pub struct InMemorySubscriber<T, S, N> {
200    transport: Arc<T>,
201    usubscription: Arc<S>,
202    notifier: Arc<N>,
203    subscription_change_listener: Arc<SubscriptionChangeListener>,
204}
205
206impl<T, S, N> core::fmt::Debug for InMemorySubscriber<T, S, N> {
207    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
208        f.debug_struct("InMemorySubscriber").finish_non_exhaustive()
209    }
210}
211
212#[cfg(feature = "communication")]
213impl<T: UTransport + 'static, P: crate::LocalUriProvider + 'static>
214    InMemorySubscriber<
215        T,
216        crate::core::usubscription::RpcClientUSubscription,
217        crate::communication::SimpleNotifier<T, P>,
218    >
219{
220    /// Creates a new Subscriber for a given transport.
221    ///
222    /// The subscriber keeps track of subscription change handlers in memory only.
223    /// This function uses the given transport to create an [`crate::core::usubscription::RpcClientUSubscription`]
224    /// and a [`crate::communication::SimpleNotifier`] and then delegate to [`Self::for_clients`]
225    /// to create the Subscriber.
226    ///
227    /// # Errors
228    ///
229    /// Returns an error if the Notifier cannot register a listener for notifications from the USubscription service.
230    pub async fn new(transport: Arc<T>, uri_provider: Arc<P>) -> Result<Self, RegistrationError> {
231        let rpc_client =
232            crate::communication::InMemoryRpcClient::new(transport.clone(), uri_provider.clone())
233                .await
234                .map(Arc::new)?;
235        let usubscription_client = Arc::new(
236            crate::core::usubscription::RpcClientUSubscription::new(rpc_client),
237        );
238        let notifier = Arc::new(crate::communication::SimpleNotifier::new(
239            transport.clone(),
240            uri_provider.clone(),
241        ));
242        Self::for_clients(transport, usubscription_client, notifier).await
243    }
244}
245
246impl<T: UTransport, S: USubscription, N: Notifier> InMemorySubscriber<T, S, N> {
247    /// Creates a new Subscriber for given clients.
248    ///
249    /// # Arguments
250    ///
251    /// * `transport` - The transport to use for registering the event listeners for subscribed topics.
252    /// * `usubscription` - The client to use for interacting with the (local) USubscription service.
253    /// * `notifier` - The client to use for registering the listener for subscription updates from USubscription.
254    ///
255    /// # Errors
256    ///
257    /// Returns an error if the Notifier cannot register a listener for notifications from the USubscription service.
258    pub async fn for_clients(
259        transport: Arc<T>,
260        usubscription: Arc<S>,
261        notifier: Arc<N>,
262    ) -> Result<Self, RegistrationError> {
263        // register a generic listener for subscription updates
264        // whenever a uE later tries to subscribe to a topic, it can provide an optional callback for
265        // handling subscription updates for the topic it tries to subscribe to
266        let subscription_change_listener = Arc::new(SubscriptionChangeListener {
267            subscription_change_handlers: RwLock::new(HashMap::new()),
268        });
269        notifier
270            .start_listening(
271                &usubscription::usubscription_uri(usubscription::RESOURCE_ID_SUBSCRIPTION_CHANGE),
272                subscription_change_listener.clone(),
273            )
274            .await?;
275        Ok(InMemorySubscriber {
276            transport,
277            usubscription,
278            notifier,
279            subscription_change_listener,
280        })
281    }
282
283    /// Stops this client.
284    ///
285    /// Clears all internal state and deregisters the listener for subscription updates from the USubscription service.
286    ///
287    /// # Errors
288    ///
289    /// Returns an error if the listener could not be unregistered. In this case the internal state remains intact.
290    pub async fn stop(&self) -> Result<(), RegistrationError> {
291        self.notifier
292            .stop_listening(
293                &usubscription::usubscription_uri(usubscription::RESOURCE_ID_SUBSCRIPTION_CHANGE),
294                self.subscription_change_listener.clone(),
295            )
296            .await
297            .and_then(|_ok| self.subscription_change_listener.clear())
298    }
299
300    async fn invoke_subscribe(
301        &self,
302        topic: &UUri,
303        subscription_change_handler: Option<Arc<dyn SubscriptionChangeHandler>>,
304    ) -> Result<SubscriptionStatus, RegistrationError> {
305        match self.usubscription.subscribe(topic, None, None).await {
306            Ok(state)
307                if state == SubscriptionStatus::Subscribed
308                    || state == SubscriptionStatus::SubscribePending =>
309            {
310                if let Some(handler) = subscription_change_handler.clone() {
311                    self.subscription_change_listener
312                        .add_handler(topic.to_owned(), handler)?;
313                }
314                Ok(state)
315            }
316            Ok(_) => {
317                debug!(topic = %topic, "failed to subscribe to topic");
318                Err(RegistrationError::Unknown(Box::from(
319                    UStatus::fail_with_code(
320                        crate::UCode::FailedPrecondition,
321                        "failed to subscribe to topic",
322                    ),
323                )))
324            }
325            Err(e) => {
326                info!(topic = %topic, "error invoking USubscription service: {}", e);
327                Err(RegistrationError::Unknown(Box::from(
328                    UStatus::fail_with_code(
329                        crate::UCode::Internal,
330                        "failed to invoke USubscription service",
331                    ),
332                )))
333            }
334        }
335    }
336
337    async fn invoke_unsubscribe(&self, topic: &UUri) -> Result<(), RegistrationError> {
338        self.usubscription
339            .unsubscribe(topic)
340            .await
341            .map(|_| {
342                let _ = self.subscription_change_listener.remove_handler(topic);
343            })
344            .map_err(|e| {
345                info!(topic = %topic, "error invoking USubscription service: {}", e);
346                RegistrationError::Unknown(Box::from(UStatus::fail_with_code(
347                    crate::UCode::Internal,
348                    "failed to invoke USubscription service",
349                )))
350            })
351    }
352
353    #[cfg(test)]
354    fn add_subscription_change_handler(
355        &self,
356        topic: &UUri,
357        subscription_change_handler: Arc<dyn SubscriptionChangeHandler>,
358    ) -> Result<(), RegistrationError> {
359        self.subscription_change_listener
360            .add_handler(topic.to_owned(), subscription_change_handler)
361    }
362
363    #[cfg(test)]
364    fn has_subscription_change_handler(&self, topic: &UUri) -> bool {
365        self.subscription_change_listener.has_handler(topic)
366    }
367}
368
369#[async_trait]
370impl<T: UTransport, U: USubscription, N: Notifier> Subscriber for InMemorySubscriber<T, U, N> {
371    async fn subscribe(
372        &self,
373        topic_filter: &UUri,
374        handler: Arc<dyn UListener>,
375        subscription_change_handler: Option<Arc<dyn SubscriptionChangeHandler>>,
376    ) -> Result<(), RegistrationError> {
377        self.invoke_subscribe(topic_filter, subscription_change_handler)
378            .await?;
379        self.transport
380            .register_listener(topic_filter, None, handler.clone())
381            .await
382            // When this fails, we have ended up in a situation where we
383            // have successfully (logically) subscribed to the topic via the USubscription service
384            // but we have not been able to register the listener with the local transport.
385            // This means that events might start getting forwarded to the local authority which
386            // are not being consumed. Apart from this inefficiency, this does not pose a real
387            // problem and since we return an err, the client might be inclined to try
388            // again and (eventually) succeed in registering the listener as well.
389            .map_err(RegistrationError::from)
390    }
391
392    async fn unsubscribe(
393        &self,
394        topic: &UUri,
395        listener: Arc<dyn UListener>,
396    ) -> Result<(), RegistrationError> {
397        self.invoke_unsubscribe(topic).await?;
398        self.transport
399            .unregister_listener(topic, None, listener)
400            .await
401            // When this fails, we have ended up in a situation where we
402            // have successfully (logically) unsubscribed from the topic via the USubscription service
403            // but we have not been able to unregister the listener from the local transport.
404            // This means that events originating from entities connected to a different transport
405            // may no longer get forwarded to the local transport, resulting in the (still registered)
406            // listener not being invoked for these events. We therefore return an error which should
407            // trigger the client to try again and (eventually) succeed in unregistering the listener as well.
408            .map_err(RegistrationError::from)
409    }
410}
411
412#[cfg(test)]
413mod tests {
414
415    // [utest->dsn~communication-layer-impl-default~1]
416
417    use super::*;
418
419    use mockall::Sequence;
420    use protobuf::well_known_types::wrappers::StringValue;
421    use usubscription::MockUSubscription;
422
423    use crate::{
424        communication::{notification::MockNotifier, pubsub::MockSubscriptionChangeHandler},
425        up_core_api::usubscription::Update,
426        utransport::{MockTransport, MockUListener},
427        UCode, UMessageBuilder, UMessageType, UStatus, UUri, UUID,
428    };
429
430    fn succeeding_notifier() -> Arc<MockNotifier> {
431        let mut notifier = MockNotifier::new();
432        notifier
433            .expect_start_listening()
434            .once()
435            .return_const(Ok(()));
436        Arc::new(notifier)
437    }
438
439    #[tokio::test]
440    async fn test_subscriber_creation_fails_when_notifier_fails_to_register_listener() {
441        // GIVEN a Notifier
442        let mut notifier = MockNotifier::new();
443        // that is not connected to its transport
444        notifier
445            .expect_start_listening()
446            .once()
447            .return_const(Err(RegistrationError::Unknown(Box::from(
448                UStatus::fail_with_code(UCode::Unavailable, "not available"),
449            ))));
450
451        // WHEN trying to create a Subscriber for this Notifier
452        let creation_attempt = InMemorySubscriber::for_clients(
453            Arc::new(MockTransport::new()),
454            Arc::new(MockUSubscription::new()),
455            Arc::new(notifier),
456        )
457        .await;
458        // THEN creation fails
459        assert!(creation_attempt.is_err_and(|e| matches!(e, RegistrationError::Unknown(_))));
460    }
461
462    #[tokio::test]
463    async fn test_subscriber_stop_succeeds() {
464        // GIVEN a Notifier
465        let mut notifier = MockNotifier::new();
466        // that succeeds to stop listening to notifications
467        notifier.expect_stop_listening().once().return_const(Ok(()));
468
469        // and a Subscriber using this Notifier
470        let subscription_change_listener = Arc::new(SubscriptionChangeListener::default());
471        let topic = UUri::try_from_parts("other", 0x1a9a, 0x01, 0x8100).unwrap();
472        let handler = Arc::new(MockSubscriptionChangeHandler::new());
473        subscription_change_listener
474            .add_handler(topic.clone(), handler)
475            .expect("adding a handler should have succeeded");
476
477        let subscriber = InMemorySubscriber {
478            transport: Arc::new(MockTransport::new()),
479            usubscription: Arc::new(MockUSubscription::new()),
480            notifier: Arc::new(notifier),
481            subscription_change_listener,
482        };
483
484        // WHEN trying to stop the Subscriber
485        let stop_attempt = subscriber.stop().await;
486
487        // THEN the attempt succeeds
488        assert!(stop_attempt.is_ok_and(|_| {
489            // and the subscription change handlers have been cleared
490            !subscriber.has_subscription_change_handler(&topic)
491        }));
492    }
493
494    #[tokio::test]
495    async fn test_subscribe_fails_when_usubscription_invocation_fails() {
496        // GIVEN a USubscription client
497        let mut seq = Sequence::new();
498        let mut usubscription_client = MockUSubscription::new();
499        // that fails to perform subscription
500        // due to different reasons
501        usubscription_client
502            .expect_subscribe()
503            .once()
504            .in_sequence(&mut seq)
505            .returning(|_, _, _| Err(UStatus::fail_with_code(UCode::Unavailable, "not connected")));
506        usubscription_client
507            .expect_subscribe()
508            .once()
509            .in_sequence(&mut seq)
510            .return_const(Ok(SubscriptionStatus::Unsubscribed));
511        // and a transport
512        let mut transport = MockTransport::new();
513        transport.expect_do_register_listener().never();
514
515        // and a Subscriber using that USubscription client
516        let subscriber = InMemorySubscriber::for_clients(
517            Arc::new(transport),
518            Arc::new(usubscription_client),
519            succeeding_notifier(),
520        )
521        .await
522        .unwrap();
523
524        // WHEN subscribing to a topic
525        let topic = UUri::try_from_parts("other", 0x1a9a, 0x01, 0x8100).unwrap();
526        let mut listener = MockUListener::new();
527        listener.expect_on_receive().never();
528        let listener_ref = Arc::new(listener);
529
530        let subscribe_attempt = subscriber
531            .subscribe(&topic, listener_ref.clone(), None)
532            .await;
533
534        // THEN the first attempt fails
535        assert!(subscribe_attempt.is_err_and(|e| matches!(e, RegistrationError::Unknown(_))));
536
537        let subscribe_attempt = subscriber
538            .subscribe(&topic, listener_ref.clone(), None)
539            .await;
540
541        // and the second attempt fails as well
542        assert!(subscribe_attempt.is_err_and(|e| matches!(e, RegistrationError::Unknown(_))));
543    }
544
545    #[tokio::test]
546    async fn test_repeated_subscribe_fails_for_different_subscription_change_handlers() {
547        // GIVEN  a USubscription client
548        let mut usubscription_client = MockUSubscription::new();
549        // that succeeds to subscribe to topics
550        usubscription_client
551            .expect_subscribe()
552            .times(2)
553            .return_const(Ok(SubscriptionStatus::Subscribed));
554
555        // and a transport
556        let mut transport = MockTransport::new();
557        // that fails to register a listener
558        transport
559            .expect_do_register_listener()
560            .once()
561            .return_const(Err(UStatus::fail_with_code(
562                UCode::Unavailable,
563                "not connected",
564            )));
565
566        // and a Subscriber using that USubscription client, Notifier and transport
567        let subscriber = InMemorySubscriber::for_clients(
568            Arc::new(transport),
569            Arc::new(usubscription_client),
570            succeeding_notifier(),
571        )
572        .await
573        .unwrap();
574
575        // WHEN subscribing to a topic
576        let topic = UUri::try_from_parts("other", 0x1a9a, 0x01, 0x8100).unwrap();
577        let listener = Arc::new(MockUListener::new());
578        let subscribe_attempt = subscriber
579            .subscribe(
580                &topic,
581                listener.clone(),
582                Some(Arc::new(MockSubscriptionChangeHandler::new())),
583            )
584            .await;
585
586        // THEN the first attempt fails due to the transport having failed
587        assert!(subscribe_attempt.is_err_and(|e| matches!(e, RegistrationError::Unknown(_))));
588
589        // and a second attempt using a different subscription change handler
590        let subscribe_attempt = subscriber
591            .subscribe(
592                &topic,
593                listener.clone(),
594                Some(Arc::new(MockSubscriptionChangeHandler::new())),
595            )
596            .await;
597        // fails with an ALREADY_EXISTS error
598        assert!(subscribe_attempt.is_err_and(|e| matches!(e, RegistrationError::AlreadyExists)));
599    }
600
601    #[tokio::test]
602    async fn test_subscribe_succeeds_on_second_attempt() {
603        let (captured_listener_tx, captured_listener_rx) = std::sync::mpsc::channel();
604
605        // GIVEN a USubscription client
606        let mut usubscription_client = MockUSubscription::new();
607        // that succeeds to subscribe to topics
608        usubscription_client
609            .expect_subscribe()
610            .times(2)
611            .return_const(Ok(SubscriptionStatus::Subscribed));
612
613        // and a transport
614        let mut transport = MockTransport::new();
615        let mut seq = Sequence::new();
616        // that first fails to register a listener
617        transport
618            .expect_do_register_listener()
619            .once()
620            .in_sequence(&mut seq)
621            .return_const(Err(UStatus::fail_with_code(
622                UCode::Unavailable,
623                "not connected",
624            )));
625        // but succeeds on the second attempt
626        transport
627            .expect_do_register_listener()
628            .once()
629            .in_sequence(&mut seq)
630            .returning(move |_source_filter, _sink_filter, listener| {
631                captured_listener_tx.send(listener).map_err(|_e| {
632                    UStatus::fail_with_code(UCode::Internal, "cannot capture listener")
633                })
634            });
635
636        // and a Subscriber using that USubscription client, Notifier and transport
637        let subscriber = InMemorySubscriber::for_clients(
638            Arc::new(transport),
639            Arc::new(usubscription_client),
640            succeeding_notifier(),
641        )
642        .await
643        .unwrap();
644
645        // WHEN subscribing to a topic
646        let topic = UUri::try_from_parts("other", 0x1a9a, 0x01, 0x8100).unwrap();
647        let mut mock_listener = MockUListener::new();
648        mock_listener.expect_on_receive().once().return_const(());
649        let listener = Arc::new(mock_listener);
650        let handler = Arc::new(MockSubscriptionChangeHandler::new());
651        let subscribe_attempt = subscriber
652            .subscribe(&topic, listener.clone(), Some(handler.clone()))
653            .await;
654
655        // THEN the first attempt fails
656        assert!(subscribe_attempt.is_err_and(|e| matches!(e, RegistrationError::Unknown(_))));
657
658        let subscribe_attempt = subscriber
659            .subscribe(&topic, listener.clone(), Some(handler.clone()))
660            .await;
661
662        // but the second attempt succeeds
663        assert!(subscribe_attempt.is_ok());
664
665        // and the registered listener receives events that are published to the topic
666        let event = UMessageBuilder::publish(topic).build().unwrap();
667        let captured_listener = captured_listener_rx.recv().unwrap().to_owned();
668        captured_listener.on_receive(event).await;
669    }
670
671    #[tokio::test]
672    async fn test_unsubscribe_fails_for_unknown_listener() {
673        // GIVEN a USubscription client
674        let mut usubscription_client = MockUSubscription::new();
675        // that succeeds to unsubscribe from topics
676        usubscription_client
677            .expect_unsubscribe()
678            .once()
679            .return_const(Ok(()));
680
681        // and a transport
682        let mut transport = MockTransport::new();
683        // which fails to unregister an unknown listener
684        transport
685            .expect_do_unregister_listener()
686            .once()
687            .return_const(Err(UStatus::fail_with_code(
688                UCode::NotFound,
689                "no such listener",
690            )));
691
692        // and a Subscriber using that USubscription client, Notifier and transport
693        let subscriber = InMemorySubscriber::for_clients(
694            Arc::new(transport),
695            Arc::new(usubscription_client),
696            succeeding_notifier(),
697        )
698        .await
699        .unwrap();
700
701        // WHEN unsubscribing from a topic for which no listener had been registered
702        let topic = UUri::try_from_parts("other", 0x1a9a, 0x01, 0x8100).unwrap();
703        let listener = Arc::new(MockUListener::new());
704        let unsubscribe_attempt = subscriber.unsubscribe(&topic, listener.clone()).await;
705
706        // THEN the the attempt fails
707        assert!(unsubscribe_attempt.is_err_and(|e| matches!(e, RegistrationError::NoSuchListener)));
708    }
709
710    #[tokio::test]
711    async fn test_unsubscribe_fails_if_usubscription_invocation_fails() {
712        // GIVEN a USubscription client
713        let mut usubscription_client = MockUSubscription::new();
714        // that fails to unsubscribe from topics
715        usubscription_client
716            .expect_unsubscribe()
717            .once()
718            .return_const(Err(UStatus::fail_with_code(UCode::Unavailable, "unknown")));
719
720        // and a transport
721        let mut transport = MockTransport::new();
722        // which succeeds to unregister listeners
723        transport
724            .expect_do_unregister_listener()
725            .never()
726            .return_const(Ok(()));
727
728        // and a Subscriber using that USubscription client, Notifier and transport
729        let subscriber = InMemorySubscriber::for_clients(
730            Arc::new(transport),
731            Arc::new(usubscription_client),
732            succeeding_notifier(),
733        )
734        .await
735        .unwrap();
736        // which already has a listener registered
737        let topic = UUri::try_from_parts("other", 0x1a9a, 0x01, 0x8100).unwrap();
738        let handler = MockSubscriptionChangeHandler::new();
739        subscriber
740            .add_subscription_change_handler(&topic, Arc::new(handler))
741            .expect("should be able to add handler");
742        assert!(subscriber.has_subscription_change_handler(&topic));
743
744        // WHEN unsubscribing from the topic
745        let listener = Arc::new(MockUListener::new());
746        let unsubscribe_attempt = subscriber.unsubscribe(&topic, listener).await;
747
748        // THEN the the attempt fails
749        assert!(unsubscribe_attempt.is_err_and(|e| {
750            matches!(e, RegistrationError::Unknown(_))
751                // and the handler is still registered
752                && subscriber.has_subscription_change_handler(&topic)
753        }));
754    }
755
756    #[tokio::test]
757    async fn test_unsubscribe_succeeds() {
758        // GIVEN a USubscription client
759        let mut usubscription_client = MockUSubscription::new();
760        // that succeeds to unsubscribe from topics
761        usubscription_client
762            .expect_unsubscribe()
763            .once()
764            .return_const(Ok(()));
765
766        // and a transport
767        let mut transport = MockTransport::new();
768        // which succeeds to unregister listeners
769        transport
770            .expect_do_unregister_listener()
771            .once()
772            .return_const(Ok(()));
773
774        // and a Subscriber using that USubscription client, Notifier and transport
775        let subscriber = InMemorySubscriber::for_clients(
776            Arc::new(transport),
777            Arc::new(usubscription_client),
778            succeeding_notifier(),
779        )
780        .await
781        .unwrap();
782        // which already has a listener registered
783        let topic = UUri::try_from_parts("other", 0x1a9a, 0x01, 0x8100).unwrap();
784        let handler = MockSubscriptionChangeHandler::new();
785        subscriber
786            .add_subscription_change_handler(&topic, Arc::new(handler))
787            .expect("should be able to add handler");
788        assert!(subscriber.has_subscription_change_handler(&topic));
789
790        // WHEN unsubscribing from a topic for which no listener had been registered
791        let listener = Arc::new(MockUListener::new());
792        let unsubscribe_attempt = subscriber.unsubscribe(&topic, listener.clone()).await;
793
794        // THEN the the attempt succeeds
795        assert!(
796            unsubscribe_attempt.is_ok_and(|_| !subscriber.has_subscription_change_handler(&topic))
797        );
798    }
799
800    #[tokio::test]
801    async fn test_unsubscribe_succeeds_on_second_attempt() {
802        // GIVEN a USubscription client
803        let mut usubscription_client = MockUSubscription::new();
804        // that succeeds to unsubscribe from topics
805        usubscription_client
806            .expect_unsubscribe()
807            .times(2)
808            .return_const(Ok(()));
809
810        // and a transport
811        let mut transport = MockTransport::new();
812        let mut seq = Sequence::new();
813        // that first fails to unregister a listener
814        transport
815            .expect_do_unregister_listener()
816            .once()
817            .in_sequence(&mut seq)
818            .return_const(Err(UStatus::fail_with_code(
819                UCode::Unavailable,
820                "not connected",
821            )));
822        // but succeeds on the second attempt
823        transport
824            .expect_do_unregister_listener()
825            .once()
826            .in_sequence(&mut seq)
827            .return_const(Ok(()));
828
829        // and a Subscriber using that USubscription client, Notifier and transport
830        let subscriber = InMemorySubscriber::for_clients(
831            Arc::new(transport),
832            Arc::new(usubscription_client),
833            succeeding_notifier(),
834        )
835        .await
836        .unwrap();
837        // which already has a listener registered
838        let topic = UUri::try_from_parts("other", 0x1a9a, 0x01, 0x8100).unwrap();
839        let handler = MockSubscriptionChangeHandler::new();
840        subscriber
841            .add_subscription_change_handler(&topic, Arc::new(handler))
842            .expect("should be able to add handler");
843        assert!(subscriber.has_subscription_change_handler(&topic));
844
845        // WHEN unsubscribing from a topic for which a listener had been registered before
846        let listener = Arc::new(MockUListener::new());
847        let unsubscribe_attempt = subscriber.unsubscribe(&topic, listener.clone()).await;
848
849        // THEN the first attempt fails
850        assert!(unsubscribe_attempt.is_err_and(|e| matches!(e, RegistrationError::Unknown(_))));
851
852        let unsubscribe_attempt = subscriber.unsubscribe(&topic, listener).await;
853
854        // but the second attempt succeeds
855        assert!(unsubscribe_attempt.is_ok_and(|_| {
856            // and the handler has been removed
857            !subscriber.has_subscription_change_handler(&topic)
858        }));
859    }
860
861    fn message_with_wrong_type(msg_type: UMessageType) -> UMessage {
862        match msg_type {
863            UMessageType::Publish => UMessageBuilder::publish(
864                UUri::try_from_parts("other", 0x1a9a, 0x01, 0x8100)
865                    .expect("should have been able to create URI"),
866            )
867            .build()
868            .expect("should have been able to create publish message"),
869            UMessageType::Notification => UMessageBuilder::notification(
870                UUri::try_from_parts("other", 0x1a9a, 0x01, 0x8100)
871                    .expect("should have been able to create origin URI"),
872                UUri::try_from_parts("other", 0x1a9b, 0x01, 0x0000)
873                    .expect("should have been able to create destination URI"),
874            )
875            .build()
876            .expect("should have been able to create notification message"),
877            UMessageType::Request => UMessageBuilder::request(
878                UUri::try_from_parts("other", 0x1a9a, 0x01, 0x0100)
879                    .expect("should have been able to create method-to-invoke URI"),
880                UUri::try_from_parts("other", 0x1a9b, 0x01, 0x0000)
881                    .expect("should have been able to create reply-to-address URI"),
882                5000,
883            )
884            .build()
885            .expect("should have been able to create request message"),
886            UMessageType::Response => UMessageBuilder::response(
887                UUri::try_from_parts("other", 0x1a9b, 0x01, 0x0000)
888                    .expect("should have been able to create reply-to-address URI"),
889                UUID::build(),
890                UUri::try_from_parts("other", 0x1a9a, 0x01, 0x0100)
891                    .expect("should have been able to create method-to-invoke URI"),
892            )
893            .build()
894            .expect("should have been able to create response message"),
895        }
896    }
897
898    fn notification_with_wrong_payload() -> UMessage {
899        UMessageBuilder::notification(
900            UUri::try_from_parts("other", 0x1a9a, 0x01, 0x8100)
901                .expect("should have been able to create origin URI"),
902            UUri::try_from_parts("other", 0x1a9b, 0x01, 0x0000)
903                .expect("should have been able to create destination URI"),
904        )
905        .build_with_protobuf_payload(&StringValue::new())
906        .expect("should have been able to create notification with payload")
907    }
908
909    fn status_update_without_topic() -> UMessage {
910        let status = crate::up_core_api::usubscription::SubscriptionStatus {
911            state: crate::up_core_api::usubscription::subscription_status::State::SUBSCRIBED.into(),
912            ..Default::default()
913        };
914        let update = Update {
915            status: Some(status).into(),
916            ..Default::default()
917        };
918        UMessageBuilder::notification(
919            UUri::try_from_parts("other", 0x1a9a, 0x01, 0x8100)
920                .expect("should have been able to create origin URI"),
921            UUri::try_from_parts("other", 0x1a9b, 0x01, 0x0000)
922                .expect("should have been able to create destination URI"),
923        )
924        .build_with_protobuf_payload(&update)
925        .expect("should have been able to create notification with payload")
926    }
927
928    fn status_update_without_status() -> UMessage {
929        let topic = UUri::try_from_parts("other", 0x1a9a, 0x01, 0x8100)
930            .expect("should have been able to create topic URI");
931        let proto_topic = crate::up_core_api::uri::UUri::from(&topic);
932        let update = Update {
933            topic: Some(proto_topic).into(),
934            ..Default::default()
935        };
936        UMessageBuilder::notification(
937            UUri::try_from_parts("other", 0x1a9a, 0x01, 0x8100)
938                .expect("should have been able to create origin URI"),
939            UUri::try_from_parts("other", 0x1a9b, 0x01, 0x0000)
940                .expect("should have been able to create destination URI"),
941        )
942        .build_with_protobuf_payload(&update)
943        .expect("should have been able to create notification with payload")
944    }
945
946    #[test_case::test_case(message_with_wrong_type(UMessageType::Publish); "Publish messages")]
947    #[test_case::test_case(message_with_wrong_type(UMessageType::Request); "Request messages")]
948    #[test_case::test_case(message_with_wrong_type(UMessageType::Response); "Response messages")]
949    #[test_case::test_case(notification_with_wrong_payload(); "wrong payload")]
950    #[test_case::test_case(status_update_without_topic(); "status without topic")]
951    #[test_case::test_case(status_update_without_status(); "update without status")]
952    #[tokio::test]
953    async fn test_subscription_change_listener_ignores(notification: UMessage) {
954        let listener = SubscriptionChangeListener::default();
955
956        let topic = UUri::try_from_parts("other", 0x1a9a, 0x01, 0x8100).unwrap();
957        let mut handler = MockSubscriptionChangeHandler::new();
958        handler.expect_on_subscription_change().never();
959
960        listener
961            .add_handler(topic.clone(), Arc::new(handler))
962            .expect("should have been able to register listener");
963        listener.on_receive(notification).await;
964    }
965
966    #[tokio::test]
967    async fn test_subscription_change_listener_invokes_handler_for_subscribed_topic() {
968        let subscriber = UUri::try_from_parts("local", 0x2000, 0x01, 0x0000)
969            .expect("should have been able to create subscriber URI");
970        let topic = UUri::try_from_parts("other", 0x1a9a, 0x01, 0x8100).unwrap();
971        let status_proto = crate::up_core_api::usubscription::SubscriptionStatus {
972            state: crate::up_core_api::usubscription::subscription_status::State::SUBSCRIBED.into(),
973            ..Default::default()
974        };
975        let update_proto = Update {
976            topic: Some((&topic).into()).into(),
977            subscriber: Some(crate::up_core_api::usubscription::SubscriberInfo {
978                uri: Some((&subscriber).into()).into(),
979                ..Default::default()
980            })
981            .into(),
982            status: Some(status_proto.clone()).into(),
983            ..Default::default()
984        };
985        let subscription_change_notification = UMessageBuilder::notification(
986            UUri::try_from_parts("local", 0x0000, 0x01, 0x8000)
987                .expect("should have been able to create uSubscription service URI"),
988            subscriber.clone(),
989        )
990        .build_with_protobuf_payload(&update_proto)
991        .expect("should have been able to create notification with payload");
992
993        let expected_topic = topic.clone();
994        let mut handler = MockSubscriptionChangeHandler::new();
995        handler
996            .expect_on_subscription_change()
997            .once()
998            .withf(move |topic, updated_status| {
999                topic == &expected_topic
1000                    && *updated_status
1001                        == SubscriptionStatus::try_from(&status_proto)
1002                            .expect("should have been able to convert status proto")
1003            })
1004            .return_const(());
1005
1006        let listener = SubscriptionChangeListener::default();
1007        listener
1008            .add_handler(topic, Arc::new(handler))
1009            .expect("should have been able to register listener");
1010
1011        listener.on_receive(subscription_change_notification).await;
1012    }
1013}
up_rust/communication/in_memory_subscriber.rs

up_rust/communication/
in_memory_subscriber.rs
