up_rust/
utransport.rs

1/********************************************************************************
2 * Copyright (c) 2023 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
14use std::fmt::{Debug, Formatter};
15use std::hash::{Hash, Hasher};
16use std::ops::Deref;
17use std::sync::Arc;
18
19#[cfg(feature = "owned-frame-transport")]
20use std::collections::HashMap;
21#[cfg(feature = "owned-frame-transport")]
22use std::sync::{LazyLock, Mutex as StdMutex};
23
24use async_trait::async_trait;
25
26#[cfg(feature = "owned-frame-transport")]
27use crate::UOwnedFrame;
28use crate::{UCode, UMessage, UStatus, UUri, UUriError};
29
30#[cfg(feature = "owned-frame-transport")]
31mod owned_transport_sealed {
32    pub trait Sealed {}
33}
34
35/// Verifies that given UUris can be used as source and sink filter UUris
36/// for registering listeners.
37///
38/// This function is helpful for implementing [`UTransport`] in accordance with the
39/// uProtocol Transport Layer specification.
40///
41/// # Errors
42///
43/// Returns a [`UStatus`] with a [`UCode::InvalidArgument`] and a corresponding detail
44/// message, if any of the given UUris cannot be used as filter criteria.
45///
46pub fn verify_filter_criteria(
47    source_filter: &UUri,
48    sink_filter: Option<&UUri>,
49) -> Result<(), Box<UStatus>> {
50    if let Some(sink_filter_uuri) = sink_filter {
51        if sink_filter_uuri.is_notification_destination()
52            && source_filter.is_notification_destination()
53        {
54            return Err(Box::from(UStatus::fail_with_code(
55                UCode::InvalidArgument,
56                "source and sink filters must not both have resource ID 0",
57            )));
58        }
59        if sink_filter_uuri.is_rpc_method()
60            && !source_filter.has_wildcard_resource_id()
61            && !source_filter.is_notification_destination()
62        {
63            return Err(Box::from(UStatus::fail_with_code(
64                UCode::InvalidArgument,
65                "source filter must either have the wildcard resource ID or resource ID 0, if sink filter matches RPC method resource ID")));
66        }
67    } else if !source_filter.has_wildcard_resource_id() && !source_filter.is_event() {
68        return Err(Box::from(UStatus::fail_with_code(
69            UCode::InvalidArgument,
70            "source filter must either have the wildcard resource ID or a resource ID from topic range, if sink filter is empty")));
71    }
72    // everything else might match valid messages
73    Ok(())
74}
75
76/// A factory for URIs representing this uEntity's resources.
77///
78/// Implementations may use arbitrary mechanisms to determine the information that
79/// is necessary for creating URIs, e.g. environment variables, configuration files etc.
80// [impl->dsn~localuriprovider-declaration~1]
81/// *Role: standalone utility answering "what is my address?"; implemented per uEntity, consumed by the roles — see the [trait map](crate::guide::trait_map).*
82///
83#[cfg_attr(any(test, feature = "test-util"), mockall::automock)]
84pub trait LocalUriProvider: Send + Sync {
85    /// Gets the _authority_ used for URIs representing this uEntity's resources.
86    fn get_authority(&self) -> String;
87    /// Gets a URI that represents a given resource of this uEntity.
88    fn get_resource_uri(&self, resource_id: u16) -> UUri;
89    /// Gets the URI that represents the resource that this uEntity expects
90    /// RPC responses and notifications to be sent to.
91    fn get_source_uri(&self) -> UUri;
92}
93
94/// A URI provider that is statically configured with the uEntity's authority, entity ID and version.
95#[derive(Debug)]
96pub struct StaticUriProvider {
97    local_uri: UUri,
98}
99
100impl StaticUriProvider {
101    /// Creates a new URI provider from static information.
102    ///
103    /// # Arguments
104    ///
105    /// * `authority` - The uEntity's authority name.
106    /// * `entity_id` - The entity identifier.
107    /// * `major_version` - The uEntity's major version.
108    ///
109    /// # Examples
110    ///
111    /// ```rust
112    /// use up_rust::{LocalUriProvider, StaticUriProvider};
113    ///
114    /// let provider = StaticUriProvider::new("my-vehicle", 0x4210, 0x05).unwrap();
115    /// assert_eq!(provider.get_authority(), "my-vehicle");
116    /// ```
117    pub fn new(
118        authority: impl Into<String>,
119        entity_id: u32,
120        major_version: u8,
121    ) -> Result<Self, UUriError> {
122        UUri::try_from_parts(authority.into().as_str(), entity_id, major_version, 0x0000)
123            .map(|local_uri| StaticUriProvider { local_uri })
124    }
125}
126
127impl LocalUriProvider for StaticUriProvider {
128    fn get_authority(&self) -> String {
129        self.local_uri.authority_name().to_owned()
130    }
131
132    fn get_resource_uri(&self, resource_id: u16) -> UUri {
133        self.local_uri.clone_with_resource_id(resource_id)
134    }
135
136    fn get_source_uri(&self) -> UUri {
137        self.local_uri.clone()
138    }
139}
140
141impl From<UUri> for StaticUriProvider {
142    fn from(value: UUri) -> Self {
143        Self::from(&value)
144    }
145}
146
147impl From<&UUri> for StaticUriProvider {
148    /// Creates a URI provider from a UUri.
149    ///
150    /// # Arguments
151    ///
152    /// * `source_uri` - The UUri to take the entity's authority, entity ID and version information from.
153    ///   The UUri's resource ID is ignored.
154    ///
155    /// # Examples
156    ///
157    /// ```rust
158    /// use up_rust::{LocalUriProvider, StaticUriProvider, UUri};
159    ///
160    /// let source_uri = UUri::try_from("//my-vehicle/4210/5/1000").unwrap();
161    /// let provider = StaticUriProvider::from(&source_uri);
162    /// assert_eq!(provider.get_authority(), "my-vehicle");
163    /// assert_eq!(provider.get_source_uri(), source_uri.clone_with_resource_id(0x0000));
164    /// ```
165    fn from(source_uri: &UUri) -> Self {
166        StaticUriProvider {
167            local_uri: source_uri.clone_with_resource_id(0x0000),
168        }
169    }
170}
171
172/// A handler for processing uProtocol messages.
173///
174/// Implementations contain the details for what should occur when a message is received.
175///
176/// Please refer to the [uProtocol Transport Layer specification](https://github.com/eclipse-uprotocol/up-spec/blob/v1.6.0-alpha.7/up-l1/README.adoc)
177/// for details.
178// [impl->dsn~ulistener-declaration~1]
179/// *Role: implemented by applications to receive `UTransport`-family [`UMessage`](crate::UMessage)s; registered on a [`UTransport`](crate::UTransport) — see the [trait map](crate::guide::trait_map).*
180///
181#[cfg_attr(any(test, feature = "test-util"), mockall::automock)]
182#[async_trait]
183pub trait UListener: Send + Sync {
184    /// Performs some action on receipt of a message.
185    ///
186    /// # Parameters
187    ///
188    /// * `msg` - The message to process.
189    ///
190    /// # Implementation hints
191    ///
192    /// This function is expected to return almost immediately. If it does not, it could potentially
193    /// block processing of succeeding messages. Long-running operations for processing a message should
194    /// therefore be run on a separate thread.
195    async fn on_receive(&self, msg: UMessage);
196}
197
198/// The uProtocol Transport Layer interface that provides a common API for uEntity developers to send and
199/// receive messages.
200///
201/// Implementations contain the details for connecting to the underlying transport technology and
202/// sending [`UMessage`]s using the configured technology.
203///
204/// Please refer to the [uProtocol Transport Layer specification](https://github.com/eclipse-uprotocol/up-spec/blob/v1.6.0-alpha.7/up-l1/README.adoc)
205/// for details.
206// [impl->dsn~utransport-declaration~1]
207/// *Role: implemented by transports; called by applications (usually via the [`communication`](crate::communication) roles) — see the [trait map](crate::guide::trait_map).*
208///
209#[async_trait]
210pub trait UTransport: Send + Sync {
211    /// Sends a message using this transport's message exchange mechanism.
212    ///
213    /// # Arguments
214    ///
215    /// * `message` - The message to send. The `type`, `source` and `sink` properties of the
216    ///   [UAttributes](https://github.com/eclipse-uprotocol/up-spec/blob/v1.6.0-alpha.7/basics/uattributes.adoc) contained
217    ///   in the message determine the addressing semantics.
218    ///
219    /// # Errors
220    ///
221    /// Returns an error if the message could not be sent.
222    async fn send(&self, message: UMessage) -> Result<(), UStatus>;
223
224    /// Receives a message from the transport.
225    ///
226    /// This default implementation returns an error with [`UCode::Unimplemented`].
227    ///
228    /// # Arguments
229    ///
230    /// * `source_filter` - The _source_ address pattern that the message to receive needs to match.
231    /// * `sink_filter` - The _sink_ address pattern that the message to receive needs to match,
232    ///                   or `None` to indicate that the message must not contain any sink address.
233    ///
234    /// # Errors
235    ///
236    /// Returns an error if no message could be received, e.g. because no message matches the given addresses.
237    async fn receive(
238        &self,
239        _source_filter: &UUri,
240        _sink_filter: Option<&UUri>,
241    ) -> Result<UMessage, UStatus> {
242        Err(UStatus::fail_with_code(
243            UCode::Unimplemented,
244            "not implemented",
245        ))
246    }
247
248    /// Registers a listener to be called for messages.
249    ///
250    /// The listener will be invoked for each message that matches the given source and sink filter patterns
251    /// according to the rules defined by the [UUri specification](https://github.com/eclipse-uprotocol/up-spec/blob/v1.6.0-alpha.7/basics/uri.adoc).
252    ///
253    /// This default implementation returns an error with [`UCode::Unimplemented`].
254    ///
255    /// # Arguments
256    ///
257    /// * `source_filter` - The _source_ address pattern that messages need to match.
258    /// * `sink_filter` - The _sink_ address pattern that messages need to match,
259    ///                   or `None` to match messages that do not contain any sink address.
260    /// * `listener` - The listener to invoke.
261    ///                The listener can be unregistered again using [`UTransport::unregister_listener`].
262    ///
263    /// # Errors
264    ///
265    /// Returns an error if the listener could not be registered.
266    async fn register_listener(
267        &self,
268        _source_filter: &UUri,
269        _sink_filter: Option<&UUri>,
270        _listener: Arc<dyn UListener>,
271    ) -> Result<(), UStatus> {
272        Err(UStatus::fail_with_code(
273            UCode::Unimplemented,
274            "not implemented",
275        ))
276    }
277
278    /// Deregisters a message listener.
279    ///
280    /// The listener will no longer be called for any (matching) messages after this function has
281    /// returned successfully.
282    ///
283    /// This default implementation returns an error with [`UCode::Unimplemented`].
284    ///
285    /// # Arguments
286    ///
287    /// * `source_filter` - The _source_ address pattern that the listener had been registered for.
288    /// * `sink_filter` - The _sink_ address pattern that the listener had been registered for.
289    /// * `listener` - The listener to unregister.
290    ///
291    /// # Errors
292    ///
293    /// Returns an error if the listener could not be unregistered, for example if the given listener does not exist.
294    async fn unregister_listener(
295        &self,
296        _source_filter: &UUri,
297        _sink_filter: Option<&UUri>,
298        _listener: Arc<dyn UListener>,
299    ) -> Result<(), UStatus> {
300        Err(UStatus::fail_with_code(
301            UCode::Unimplemented,
302            "not implemented",
303        ))
304    }
305}
306
307#[cfg(feature = "owned-frame-transport")]
308fn verify_owned_filter_criteria(
309    source_filter: &UUri,
310    sink_filter: Option<&UUri>,
311) -> Result<(), UStatus> {
312    verify_filter_criteria(source_filter, sink_filter).map_err(|status| *status)
313}
314
315#[cfg(feature = "owned-frame-transport")]
316#[derive(Clone, Debug, Eq, Hash, PartialEq)]
317struct OwnedListenerRegistrationKey {
318    transport: usize,
319    source_filter: UUri,
320    sink_filter: Option<UUri>,
321    listener: usize,
322}
323
324#[cfg(feature = "owned-frame-transport")]
325static OWNED_LISTENER_REGISTRY: LazyLock<
326    StdMutex<HashMap<OwnedListenerRegistrationKey, Arc<dyn UOwnedListener>>>,
327> = LazyLock::new(|| StdMutex::new(HashMap::new()));
328
329#[cfg(feature = "owned-frame-transport")]
330fn owned_transport_pointer<T: ?Sized>(transport: &T) -> usize {
331    let ptr = transport as *const T;
332    let thin_ptr = ptr as *const ();
333    thin_ptr as usize
334}
335
336#[cfg(feature = "owned-frame-transport")]
337fn owned_listener_pointer(listener: &Arc<dyn UOwnedListener>) -> usize {
338    let ptr = Arc::as_ptr(listener);
339    let thin_ptr = ptr as *const ();
340    thin_ptr as usize
341}
342
343#[cfg(feature = "owned-frame-transport")]
344fn owned_listener_registration_key<T: ?Sized>(
345    transport: &T,
346    source_filter: &UUri,
347    sink_filter: Option<&UUri>,
348    listener: usize,
349) -> OwnedListenerRegistrationKey {
350    OwnedListenerRegistrationKey {
351        transport: owned_transport_pointer(transport),
352        source_filter: source_filter.clone(),
353        sink_filter: sink_filter.cloned(),
354        listener,
355    }
356}
357
358// Delivered frames are `UOwnedFrame<Validated>` by type; no re-validation
359// wrapper is needed here.
360
361#[cfg(feature = "owned-frame-transport")]
362fn registered_owned_listener(
363    key: &OwnedListenerRegistrationKey,
364    listener: Arc<dyn UOwnedListener>,
365) -> (Arc<dyn UOwnedListener>, bool) {
366    let mut registry = OWNED_LISTENER_REGISTRY
367        .lock()
368        .expect("owned listener registry lock poisoned");
369    if let Some(existing) = registry.get(key) {
370        return (existing.clone(), false);
371    }
372
373    registry.insert(key.clone(), listener.clone());
374    (listener, true)
375}
376
377#[cfg(feature = "owned-frame-transport")]
378fn owned_listener_for_unregister(
379    key: &OwnedListenerRegistrationKey,
380    fallback: Arc<dyn UOwnedListener>,
381) -> Arc<dyn UOwnedListener> {
382    OWNED_LISTENER_REGISTRY
383        .lock()
384        .expect("owned listener registry lock poisoned")
385        .get(key)
386        .cloned()
387        .unwrap_or(fallback)
388}
389
390/// *Role: implemented by applications to receive owned frames — see the [trait map](crate::guide::trait_map).*
391///
392/// Listener for experimental owned native-frame transports.
393#[cfg(feature = "owned-frame-transport")]
394#[cfg_attr(any(test, feature = "test-util"), mockall::automock)]
395#[async_trait]
396pub trait UOwnedListener: Send + Sync {
397    /// Performs some action on receipt of an owned native frame.
398    async fn on_receive_owned(&self, frame: UOwnedFrame);
399}
400
401/// *Role: implemented by transports offering the owned-frame family; users call [`UOwnedTransport`](crate::UOwnedTransport) — see the [trait map](crate::guide::trait_map).*
402///
403/// *Role: called by users of the owned-frame family; transports implement [`UOwnedTransportImpl`](crate::UOwnedTransportImpl) instead — see the [trait map](crate::guide::trait_map).*
404///
405/// Implementation boundary for experimental owned native-frame transports.
406///
407/// Implementing this trait buys the public [`UOwnedTransport`] API, frame and
408/// filter validation, listener validation, and compatibility projections above
409/// the transport. Only the [`Validated`](crate::Validated) frame state reaches
410/// the required send method — constructors validate, and the only path from
411/// [`Unvalidated`](crate::Unvalidated) is [`UOwnedFrame::validate`] — so
412/// `req~owned-frame-validate-before-send~1` is enforced by the type system
413/// before delegation.
414///
415/// Send is required. Pull receive and listener registration have default
416/// unsupported implementations and are overridden only for carriage patterns
417/// the technology supports. `UOwnedTransportCore` is the different,
418/// selected-wire seam for transports that carry encoded metadata bytes.
419#[cfg(feature = "owned-frame-transport")]
420#[async_trait]
421pub trait UOwnedTransportImpl: Send + Sync {
422    /// Sends an already validated owned frame.
423    async fn send_validated_owned(&self, frame: UOwnedFrame) -> Result<(), UStatus>;
424
425    /// Receives one matching owned frame from transports that support pull receive.
426    async fn receive_validated_owned(
427        &self,
428        _source_filter: &UUri,
429        _sink_filter: Option<&UUri>,
430    ) -> Result<UOwnedFrame, UStatus> {
431        Err(UStatus::fail_with_code(
432            UCode::Unimplemented,
433            "not implemented",
434        ))
435    }
436
437    /// Registers an owned listener after public filter validation.
438    async fn register_validated_owned_listener(
439        &self,
440        _source_filter: &UUri,
441        _sink_filter: Option<&UUri>,
442        _listener: Arc<dyn UOwnedListener>,
443    ) -> Result<(), UStatus> {
444        Err(UStatus::fail_with_code(
445            UCode::Unimplemented,
446            "not implemented",
447        ))
448    }
449
450    /// Unregisters an owned listener after public filter validation.
451    async fn unregister_validated_owned_listener(
452        &self,
453        _source_filter: &UUri,
454        _sink_filter: Option<&UUri>,
455        _listener: Arc<dyn UOwnedListener>,
456    ) -> Result<(), UStatus> {
457        Err(UStatus::fail_with_code(
458            UCode::Unimplemented,
459            "not implemented",
460        ))
461    }
462}
463
464#[cfg(feature = "owned-frame-transport")]
465impl<T> owned_transport_sealed::Sealed for T where T: UOwnedTransportImpl + ?Sized {}
466
467/// Experimental serialization-neutral owned native-frame transport API.
468///
469/// This API is additive to [`UTransport`]. It does not replace the ordinary
470/// `UMessage` compatibility path and intentionally does not include copying
471/// adapters between transport families.
472#[cfg(feature = "owned-frame-transport")]
473#[async_trait]
474pub trait UOwnedTransport: owned_transport_sealed::Sealed + Send + Sync {
475    /// Sends an owned native frame after public validation.
476    async fn send_owned(&self, frame: UOwnedFrame) -> Result<(), UStatus>;
477
478    /// Receives one matching owned frame from transports that support pull receive.
479    async fn receive_owned(
480        &self,
481        source_filter: &UUri,
482        sink_filter: Option<&UUri>,
483    ) -> Result<UOwnedFrame, UStatus>;
484
485    /// Registers a listener for matching owned native frames.
486    async fn register_owned_listener(
487        &self,
488        source_filter: &UUri,
489        sink_filter: Option<&UUri>,
490        listener: Arc<dyn UOwnedListener>,
491    ) -> Result<(), UStatus>;
492
493    /// Unregisters a listener for matching owned native frames.
494    async fn unregister_owned_listener(
495        &self,
496        source_filter: &UUri,
497        sink_filter: Option<&UUri>,
498        listener: Arc<dyn UOwnedListener>,
499    ) -> Result<(), UStatus>;
500}
501
502#[cfg(feature = "owned-frame-transport")]
503#[async_trait]
504impl<T> UOwnedTransport for T
505where
506    T: UOwnedTransportImpl + ?Sized,
507{
508    async fn send_owned(&self, frame: UOwnedFrame) -> Result<(), UStatus> {
509        self.send_validated_owned(frame).await
510    }
511
512    async fn receive_owned(
513        &self,
514        source_filter: &UUri,
515        sink_filter: Option<&UUri>,
516    ) -> Result<UOwnedFrame, UStatus> {
517        verify_owned_filter_criteria(source_filter, sink_filter)?;
518        let frame =
519            UOwnedTransportImpl::receive_validated_owned(self, source_filter, sink_filter).await?;
520        Ok(frame)
521    }
522
523    async fn register_owned_listener(
524        &self,
525        source_filter: &UUri,
526        sink_filter: Option<&UUri>,
527        listener: Arc<dyn UOwnedListener>,
528    ) -> Result<(), UStatus> {
529        verify_owned_filter_criteria(source_filter, sink_filter)?;
530        let key = owned_listener_registration_key(
531            self,
532            source_filter,
533            sink_filter,
534            owned_listener_pointer(&listener),
535        );
536        let (listener, inserted) = registered_owned_listener(&key, listener);
537        let result = UOwnedTransportImpl::register_validated_owned_listener(
538            self,
539            source_filter,
540            sink_filter,
541            listener,
542        )
543        .await;
544        if result.is_err() && inserted {
545            OWNED_LISTENER_REGISTRY
546                .lock()
547                .expect("owned listener registry lock poisoned")
548                .remove(&key);
549        }
550        result
551    }
552
553    async fn unregister_owned_listener(
554        &self,
555        source_filter: &UUri,
556        sink_filter: Option<&UUri>,
557        listener: Arc<dyn UOwnedListener>,
558    ) -> Result<(), UStatus> {
559        verify_owned_filter_criteria(source_filter, sink_filter)?;
560        let key = owned_listener_registration_key(
561            self,
562            source_filter,
563            sink_filter,
564            owned_listener_pointer(&listener),
565        );
566        let listener = owned_listener_for_unregister(&key, listener);
567        let result = UOwnedTransportImpl::unregister_validated_owned_listener(
568            self,
569            source_filter,
570            sink_filter,
571            listener,
572        )
573        .await;
574        if result.is_ok() {
575            OWNED_LISTENER_REGISTRY
576                .lock()
577                .expect("owned listener registry lock poisoned")
578                .remove(&key);
579        }
580        result
581    }
582}
583
584#[cfg(not(tarpaulin_include))]
585#[cfg(all(feature = "owned-frame-transport", any(test, feature = "test-util")))]
586mockall::mock! {
587    /// Mockall-generated mock for test-util consumers.
588    pub UOwnedTransport {
589        /// Mock hook; configure with mockall expectations.
590    ///
591    /// # Errors
592    ///
593    /// Returns whatever the test's expectations are configured to return.
594        pub async fn do_send_validated_owned(&self, frame: UOwnedFrame) -> Result<(), UStatus>;
595        /// Mock hook; configure with mockall expectations.
596    ///
597    /// # Errors
598    ///
599    /// Returns whatever the test's expectations are configured to return.
600        pub async fn do_receive_validated_owned<'a>(&'a self, source_filter: &'a UUri, sink_filter: Option<&'a UUri>) -> Result<UOwnedFrame, UStatus>;
601        /// Mock hook; configure with mockall expectations.
602    ///
603    /// # Errors
604    ///
605    /// Returns whatever the test's expectations are configured to return.
606        pub async fn do_register_validated_owned_listener<'a>(&'a self, source_filter: &'a UUri, sink_filter: Option<&'a UUri>, listener: Arc<dyn UOwnedListener>) -> Result<(), UStatus>;
607        /// Mock hook; configure with mockall expectations.
608    ///
609    /// # Errors
610    ///
611    /// Returns whatever the test's expectations are configured to return.
612        pub async fn do_unregister_validated_owned_listener<'a>(&'a self, source_filter: &'a UUri, sink_filter: Option<&'a UUri>, listener: Arc<dyn UOwnedListener>) -> Result<(), UStatus>;
613    }
614}
615
616#[cfg(not(tarpaulin_include))]
617#[cfg(all(feature = "owned-frame-transport", any(test, feature = "test-util")))]
618#[async_trait]
619impl UOwnedTransportImpl for MockUOwnedTransport {
620    async fn send_validated_owned(&self, frame: UOwnedFrame) -> Result<(), UStatus> {
621        self.do_send_validated_owned(frame).await
622    }
623
624    async fn receive_validated_owned(
625        &self,
626        source_filter: &UUri,
627        sink_filter: Option<&UUri>,
628    ) -> Result<UOwnedFrame, UStatus> {
629        self.do_receive_validated_owned(source_filter, sink_filter)
630            .await
631    }
632
633    async fn register_validated_owned_listener(
634        &self,
635        source_filter: &UUri,
636        sink_filter: Option<&UUri>,
637        listener: Arc<dyn UOwnedListener>,
638    ) -> Result<(), UStatus> {
639        self.do_register_validated_owned_listener(source_filter, sink_filter, listener)
640            .await
641    }
642
643    async fn unregister_validated_owned_listener(
644        &self,
645        source_filter: &UUri,
646        sink_filter: Option<&UUri>,
647        listener: Arc<dyn UOwnedListener>,
648    ) -> Result<(), UStatus> {
649        self.do_unregister_validated_owned_listener(source_filter, sink_filter, listener)
650            .await
651    }
652}
653
654#[cfg(not(tarpaulin_include))]
655#[cfg(any(test, feature = "test-util"))]
656mockall::mock! {
657    /// This extra struct is necessary in order to comply with mockall's requirements regarding the parameter lifetimes
658    /// see <https://github.com/asomers/mockall/issues/571>
659    pub Transport {
660        /// Mock hook; configure with mockall expectations.
661    ///
662    /// # Errors
663    ///
664    /// Returns whatever the test's expectations are configured to return.
665        pub async fn do_send(&self, message: UMessage) -> Result<(), UStatus>;
666        /// Mock hook; configure with mockall expectations.
667    ///
668    /// # Errors
669    ///
670    /// Returns whatever the test's expectations are configured to return.
671        pub async fn do_register_listener<'a>(&'a self, source_filter: &'a UUri, sink_filter: Option<&'a UUri>, listener: Arc<dyn UListener>) -> Result<(), UStatus>;
672        /// Mock hook; configure with mockall expectations.
673    ///
674    /// # Errors
675    ///
676    /// Returns whatever the test's expectations are configured to return.
677        pub async fn do_unregister_listener<'a>(&'a self, source_filter: &'a UUri, sink_filter: Option<&'a UUri>, listener: Arc<dyn UListener>) -> Result<(), UStatus>;
678    }
679}
680
681#[cfg(not(tarpaulin_include))]
682#[cfg(any(test, feature = "test-util"))]
683#[async_trait]
684/// This delegates the invocation of the UTransport functions to the mocked functions of the Transport struct.
685/// see <https://github.com/asomers/mockall/issues/571>
686impl UTransport for MockTransport {
687    async fn send(&self, message: UMessage) -> Result<(), UStatus> {
688        self.do_send(message).await
689    }
690    async fn register_listener(
691        &self,
692        source_filter: &UUri,
693        sink_filter: Option<&UUri>,
694        listener: Arc<dyn UListener>,
695    ) -> Result<(), UStatus> {
696        self.do_register_listener(source_filter, sink_filter, listener)
697            .await
698    }
699    async fn unregister_listener(
700        &self,
701        source_filter: &UUri,
702        sink_filter: Option<&UUri>,
703        listener: Arc<dyn UListener>,
704    ) -> Result<(), UStatus> {
705        self.do_unregister_listener(source_filter, sink_filter, listener)
706            .await
707    }
708}
709
710/// A wrapper type that allows comparing [`UListener`]s to each other.
711///
712/// # Note
713///
714/// Not necessary for end-user uEs to use. Primarily intended for `up-client-foo-rust` UPClient libraries
715/// when implementing [`UTransport`].
716///
717/// # Rationale
718///
719/// The wrapper type is implemented such that it can be used in any location you may wish to
720/// hold a type implementing [`UListener`].
721///
722/// Implements necessary traits to allow hashing, so that you may hold the wrapper type in
723/// collections which require that, such as a `HashMap` or `HashSet`
724#[derive(Clone)]
725pub struct ComparableListener {
726    listener: Arc<dyn UListener>,
727}
728
729impl ComparableListener {
730    /// Creates a comparable wrapper around a listener for registry equality.
731    pub fn new(listener: Arc<dyn UListener>) -> Self {
732        Self { listener }
733    }
734    /// Gets a clone of the wrapped reference to the listener.
735    #[must_use]
736    pub fn into_inner(&self) -> Arc<dyn UListener> {
737        self.listener.clone()
738    }
739
740    /// Allows us to get the pointer address of this `ComparableListener` on the heap
741    fn pointer_address(&self) -> usize {
742        // Obtain the raw pointer from the Arc
743        let ptr = Arc::as_ptr(&self.listener);
744        // Cast the fat pointer to a raw thin pointer to ()
745        let thin_ptr = ptr as *const ();
746        // Convert the thin pointer to a usize
747        thin_ptr as usize
748    }
749}
750
751impl Deref for ComparableListener {
752    type Target = dyn UListener;
753
754    fn deref(&self) -> &Self::Target {
755        &*self.listener
756    }
757}
758
759impl Hash for ComparableListener {
760    /// Feeds the pointer to the listener held by `self` into the given [`Hasher`].
761    ///
762    /// This is consistent with the implementation of [`ComparableListener::eq`].
763    fn hash<H: Hasher>(&self, state: &mut H) {
764        Arc::as_ptr(&self.listener).hash(state);
765    }
766}
767
768impl PartialEq for ComparableListener {
769    /// Compares this listener to another listener.
770    ///
771    /// # Returns
772    ///
773    /// `true` if the pointer to the listener held by `self` is equal to the pointer held by `other`.
774    /// This is consistent with the implementation of [`ComparableListener::hash`].
775    fn eq(&self, other: &Self) -> bool {
776        Arc::ptr_eq(&self.listener, &other.listener)
777    }
778}
779
780impl Eq for ComparableListener {}
781
782impl Debug for ComparableListener {
783    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
784        write!(f, "ComparableListener: {}", self.pointer_address())
785    }
786}
787
788#[cfg(test)]
789mod tests {
790    use crate::{ComparableListener, UListener, UMessage, UMessageBuilder};
791    use std::{
792        hash::{DefaultHasher, Hash, Hasher},
793        ops::Deref,
794        str::FromStr,
795        sync::Arc,
796    };
797
798    use super::*;
799
800    #[test]
801    fn test_static_uri_provider_get_source() {
802        let provider = StaticUriProvider::new("my-vehicle", 0x4210, 0x05)
803            .expect("failed to create URI provider");
804        let source_uri = provider.get_source_uri();
805        assert_eq!(source_uri.authority_name(), "my-vehicle");
806        assert_eq!(source_uri.uentity_type_id(), 0x4210);
807        assert_eq!(source_uri.uentity_major_version(), 0x05);
808        assert_eq!(source_uri.resource_id(), 0x0000);
809    }
810
811    #[test]
812    fn test_static_uri_provider_get_resource() {
813        let provider = StaticUriProvider::new("my-vehicle", 0x4210, 0x05)
814            .expect("failed to create URI provider");
815        let resource_uri = provider.get_resource_uri(0x1234);
816        assert_eq!(resource_uri.authority_name(), "my-vehicle");
817        assert_eq!(resource_uri.uentity_type_id(), 0x4210);
818        assert_eq!(resource_uri.uentity_major_version(), 0x05);
819        assert_eq!(resource_uri.resource_id(), 0x1234);
820    }
821
822    #[tokio::test]
823    async fn test_deref_returns_wrapped_listener() {
824        let empty_message = UMessageBuilder::publish(
825            UUri::try_from_parts("my-vehicle", 0x4210, 0x05, 0x9000)
826                .expect("failed to create topic"),
827        )
828        .build()
829        .expect("failed to build message");
830        let mut mock_listener = MockUListener::new();
831        mock_listener.expect_on_receive().once().return_const(());
832        let listener_one = Arc::new(mock_listener);
833        let comparable_listener_one = ComparableListener::new(listener_one);
834        comparable_listener_one
835            .deref()
836            .on_receive(empty_message)
837            .await;
838    }
839
840    #[tokio::test]
841    async fn test_to_inner_returns_reference_to_wrapped_listener() {
842        let empty_message = UMessageBuilder::publish(
843            UUri::try_from_parts("my-vehicle", 0x4210, 0x05, 0x9000)
844                .expect("failed to create topic"),
845        )
846        .build()
847        .expect("failed to build message");
848        let mut mock_listener = MockUListener::new();
849        mock_listener.expect_on_receive().once().return_const(());
850        let listener_one = Arc::new(mock_listener);
851        let comparable_listener_one = ComparableListener::new(listener_one);
852        comparable_listener_one
853            .into_inner()
854            .on_receive(empty_message)
855            .await;
856    }
857
858    #[tokio::test]
859    async fn test_eq_and_hash_are_consistent_for_comparable_listeners_wrapping_same_listener() {
860        let empty_message = UMessageBuilder::publish(
861            UUri::try_from_parts("my-vehicle", 0x4210, 0x05, 0x9000)
862                .expect("failed to create topic"),
863        )
864        .build()
865        .expect("failed to build message");
866        let mut mock_listener = MockUListener::new();
867        mock_listener.expect_on_receive().times(2).return_const(());
868        let listener_one = Arc::new(mock_listener);
869        let listener_two = listener_one.clone();
870        listener_one.on_receive(empty_message.clone()).await;
871        listener_two.on_receive(empty_message.clone()).await;
872        let comparable_listener_one = ComparableListener::new(listener_one);
873        let comparable_listener_two = ComparableListener::new(listener_two);
874        assert!(&comparable_listener_one.eq(&comparable_listener_two));
875
876        let mut hasher = DefaultHasher::new();
877        comparable_listener_one.hash(&mut hasher);
878        let hash_one = hasher.finish();
879        let mut hasher = DefaultHasher::new();
880        comparable_listener_two.hash(&mut hasher);
881        let hash_two = hasher.finish();
882        assert_eq!(hash_one, hash_two);
883    }
884
885    #[tokio::test]
886    async fn test_eq_and_hash_are_consistent_for_comparable_listeners_wrapping_different_listeners()
887    {
888        let empty_message = UMessageBuilder::publish(
889            UUri::try_from_parts("my-vehicle", 0x4210, 0x05, 0x9000)
890                .expect("failed to create topic"),
891        )
892        .build()
893        .expect("failed to build message");
894
895        let mut mock_listener_one = MockUListener::new();
896        mock_listener_one
897            .expect_on_receive()
898            .once()
899            .return_const(());
900        let listener_one = Arc::new(mock_listener_one);
901        let mut mock_listener_two = MockUListener::new();
902        mock_listener_two
903            .expect_on_receive()
904            .once()
905            .return_const(());
906        let listener_two = Arc::new(mock_listener_two);
907        listener_one.on_receive(empty_message.clone()).await;
908        listener_two.on_receive(empty_message.clone()).await;
909        let comparable_listener_one = ComparableListener::new(listener_one);
910        let comparable_listener_two = ComparableListener::new(listener_two);
911        assert!(!&comparable_listener_one.eq(&comparable_listener_two));
912
913        let mut hasher = DefaultHasher::new();
914        comparable_listener_one.hash(&mut hasher);
915        let hash_one = hasher.finish();
916        let mut hasher = DefaultHasher::new();
917        comparable_listener_two.hash(&mut hasher);
918        let hash_two = hasher.finish();
919        assert_ne!(hash_one, hash_two);
920    }
921
922    #[tokio::test]
923    async fn test_utransport_default_implementations() {
924        struct EmptyTransport {}
925        #[async_trait::async_trait]
926        impl UTransport for EmptyTransport {
927            async fn send(&self, _message: UMessage) -> Result<(), UStatus> {
928                todo!()
929            }
930        }
931
932        let transport = EmptyTransport {};
933        let listener = Arc::new(MockUListener::new());
934
935        assert!(transport
936            .receive(&UUri::any(), None)
937            .await
938            .is_err_and(|e| e.code() == UCode::Unimplemented));
939        assert!(transport
940            .register_listener(&UUri::any(), None, listener.clone())
941            .await
942            .is_err_and(|e| e.code() == UCode::Unimplemented));
943        assert!(transport
944            .unregister_listener(&UUri::any(), None, listener)
945            .await
946            .is_err_and(|e| e.code() == UCode::Unimplemented));
947    }
948
949    #[test]
950    fn test_comparable_listener_pointer_address() {
951        let bar = Arc::new(MockUListener::new());
952        let comp_listener = ComparableListener::new(bar);
953
954        let comp_listener_thread = comp_listener.clone();
955        let handle = std::thread::spawn(move || comp_listener_thread.pointer_address());
956
957        let comp_listener_address_other_thread = handle.join().unwrap();
958        let comp_listener_address_this_thread = comp_listener.pointer_address();
959
960        assert_eq!(
961            comp_listener_address_this_thread,
962            comp_listener_address_other_thread
963        );
964    }
965
966    #[test]
967    fn test_comparable_listener_debug_outputs() {
968        let bar = Arc::new(MockUListener::new());
969        let comp_listener = ComparableListener::new(bar);
970        let debug_output = format!("{comp_listener:?}");
971        assert!(!debug_output.is_empty());
972    }
973
974    #[test_case::test_case(
975        "//vehicle1/AA/1/FFFF",
976        Some("//vehicle2/BB/1/FFFF");
977        "source and sink both having wildcard resource ID")]
978    #[test_case::test_case(
979        "//vehicle1/AA/1/9000",
980        Some("//vehicle2/BB/1/0");
981        "sending notification")]
982    #[test_case::test_case(
983        "//vehicle1/AA/1/0",
984        Some("//vehicle2/BB/1/1");
985        "RPC method invocation")]
986    #[test_case::test_case(
987        "//vehicle1/AA/1/FFFF",
988        Some("//vehicle2/BB/1/1");
989        "receiving RPC requests using wildcard resource ID")]
990    #[test_case::test_case(
991        "//vehicle1/AA/1/0",
992        Some("//vehicle2/BB/1/1");
993        "receiving RPC requests using default resource ID")]
994    #[test_case::test_case(
995        "//vehicle1/AA/1/9000",
996        None;
997        "receiving events published to specific topic")]
998    #[test_case::test_case(
999        "//vehicle1/AA/1/FFFF",
1000        None;
1001        "receiving events published to any topic")]
1002    fn test_verify_filter_criteria_succeeds_for(source: &str, sink: Option<&str>) {
1003        let source_filter = UUri::from_str(source).expect("invalid source URI");
1004        let sink_filter = sink.map(|s| UUri::from_str(s).expect("invalid sink URI"));
1005        assert!(verify_filter_criteria(&source_filter, sink_filter.as_ref()).is_ok());
1006    }
1007
1008    #[test_case::test_case(
1009        UUri::from_str("//vehicle1/AA/1/0").unwrap(),
1010        Some(UUri::from_str("//vehicle2/BB/1/0").unwrap());
1011        "source and sink both having resource ID 0")]
1012    #[test_case::test_case(
1013        UUri::from_str("//vehicle1/AA/1/CC").unwrap(),
1014        Some(UUri::from_str("//vehicle2/BB/1/1A").unwrap());
1015        "sink is RPC but source has invalid resource ID")]
1016    #[test_case::test_case(
1017        UUri::from_str("//vehicle1/AA/1/CC").unwrap(),
1018        None;
1019        "sink is empty but source has non-topic resource ID")]
1020    fn test_verify_filter_criteria_fails_for(source_filter: UUri, sink_filter: Option<UUri>) {
1021        assert!(verify_filter_criteria(&source_filter, sink_filter.as_ref())
1022            .is_err_and(|err| matches!(err.code(), UCode::InvalidArgument)));
1023    }
1024}
1025
1026#[cfg(all(test, feature = "owned-frame-transport"))]
1027mod owned_transport_tests {
1028    use std::str::FromStr;
1029    use std::sync::Mutex;
1030
1031    use async_trait::async_trait;
1032    use bytes::Bytes;
1033
1034    use super::*;
1035    use crate::{PayloadEncoding, UFrameMetadata, UMessageBuilder};
1036
1037    fn topic() -> UUri {
1038        UUri::try_from_parts("vehicle", 0x4210, 0x01, 0x9000).expect("failed to create topic")
1039    }
1040
1041    fn invalid_source_filter() -> UUri {
1042        UUri::from_str("//vehicle/4210/1/10").expect("invalid test URI")
1043    }
1044
1045    fn metadata_with_encoding() -> UFrameMetadata {
1046        let message = UMessageBuilder::publish(topic()).build().expect("message");
1047        crate::frame::metadata::try_project_attributes_to_frame_metadata(
1048            message.attributes(),
1049            Some(PayloadEncoding::RAW),
1050        )
1051        .expect("metadata")
1052    }
1053
1054    fn valid_frame() -> UOwnedFrame {
1055        UOwnedFrame::with_payload(metadata_with_encoding(), Bytes::new()).expect("valid frame")
1056    }
1057
1058    #[tokio::test]
1059    async fn owned_transport_defaults_are_unimplemented() {
1060        struct EmptyTransport;
1061
1062        #[async_trait]
1063        impl UOwnedTransportImpl for EmptyTransport {
1064            async fn send_validated_owned(&self, _frame: UOwnedFrame) -> Result<(), UStatus> {
1065                Ok(())
1066            }
1067        }
1068
1069        let transport = EmptyTransport;
1070        let listener = Arc::new(MockUOwnedListener::new());
1071        let source = UUri::any();
1072
1073        assert!(transport
1074            .receive_owned(&source, None)
1075            .await
1076            .is_err_and(|status| status.code() == UCode::Unimplemented));
1077        assert!(transport
1078            .register_owned_listener(&source, None, listener.clone())
1079            .await
1080            .is_err_and(|status| status.code() == UCode::Unimplemented));
1081        assert!(transport
1082            .unregister_owned_listener(&source, None, listener)
1083            .await
1084            .is_err_and(|status| status.code() == UCode::Unimplemented));
1085    }
1086
1087    // Sending an unvalidated frame is a compile error (`send_owned` takes
1088    // `UOwnedFrame<Validated>`); the pin for that lives in the trybuild battery.
1089
1090    #[tokio::test]
1091    async fn owned_transport_validates_filters_before_receive_impl() {
1092        #[derive(Default)]
1093        struct ReceiveCountingTransport {
1094            receive_count: Mutex<usize>,
1095        }
1096
1097        #[async_trait]
1098        impl UOwnedTransportImpl for ReceiveCountingTransport {
1099            async fn send_validated_owned(&self, _frame: UOwnedFrame) -> Result<(), UStatus> {
1100                Ok(())
1101            }
1102
1103            async fn receive_validated_owned(
1104                &self,
1105                _source_filter: &UUri,
1106                _sink_filter: Option<&UUri>,
1107            ) -> Result<UOwnedFrame, UStatus> {
1108                *self
1109                    .receive_count
1110                    .lock()
1111                    .expect("receive count lock poisoned") += 1;
1112                Ok(valid_frame())
1113            }
1114        }
1115
1116        let transport = ReceiveCountingTransport::default();
1117        let status = transport
1118            .receive_owned(&invalid_source_filter(), None)
1119            .await
1120            .expect_err("invalid filters must be rejected");
1121
1122        assert_eq!(status.code(), UCode::InvalidArgument);
1123        assert_eq!(*transport.receive_count.lock().unwrap(), 0);
1124    }
1125
1126    #[tokio::test]
1127    async fn owned_transport_validates_filters_before_register_impl() {
1128        #[derive(Default)]
1129        struct RegisterCountingTransport {
1130            register_count: Mutex<usize>,
1131        }
1132
1133        #[async_trait]
1134        impl UOwnedTransportImpl for RegisterCountingTransport {
1135            async fn send_validated_owned(&self, _frame: UOwnedFrame) -> Result<(), UStatus> {
1136                Ok(())
1137            }
1138
1139            async fn register_validated_owned_listener(
1140                &self,
1141                _source_filter: &UUri,
1142                _sink_filter: Option<&UUri>,
1143                _listener: Arc<dyn UOwnedListener>,
1144            ) -> Result<(), UStatus> {
1145                *self
1146                    .register_count
1147                    .lock()
1148                    .expect("register count lock poisoned") += 1;
1149                Ok(())
1150            }
1151        }
1152
1153        let transport = RegisterCountingTransport::default();
1154        let listener = Arc::new(MockUOwnedListener::new());
1155        let status = transport
1156            .register_owned_listener(&invalid_source_filter(), None, listener)
1157            .await
1158            .expect_err("invalid filters must be rejected");
1159
1160        assert_eq!(status.code(), UCode::InvalidArgument);
1161        assert_eq!(*transport.register_count.lock().unwrap(), 0);
1162    }
1163
1164    #[derive(Default)]
1165    struct CountingOwnedListener {
1166        count: Mutex<usize>,
1167    }
1168
1169    impl CountingOwnedListener {
1170        fn count(&self) -> usize {
1171            *self
1172                .count
1173                .lock()
1174                .expect("owned listener count lock poisoned")
1175        }
1176    }
1177
1178    #[async_trait]
1179    impl UOwnedListener for CountingOwnedListener {
1180        async fn on_receive_owned(&self, _frame: UOwnedFrame) {
1181            *self
1182                .count
1183                .lock()
1184                .expect("owned listener count lock poisoned") += 1;
1185        }
1186    }
1187
1188    #[derive(Default)]
1189    struct ListenerSpyOwnedTransport {
1190        listener: Mutex<Option<Arc<dyn UOwnedListener>>>,
1191    }
1192
1193    #[async_trait]
1194    impl UOwnedTransportImpl for ListenerSpyOwnedTransport {
1195        async fn send_validated_owned(&self, _frame: UOwnedFrame) -> Result<(), UStatus> {
1196            Ok(())
1197        }
1198
1199        async fn register_validated_owned_listener(
1200            &self,
1201            _source_filter: &UUri,
1202            _sink_filter: Option<&UUri>,
1203            listener: Arc<dyn UOwnedListener>,
1204        ) -> Result<(), UStatus> {
1205            *self.listener.lock().expect("listener lock poisoned") = Some(listener);
1206            Ok(())
1207        }
1208    }
1209
1210    #[tokio::test]
1211    async fn validating_owned_listener_drops_invalid_frames_before_user_callback() {
1212        let transport = ListenerSpyOwnedTransport::default();
1213        let listener = Arc::new(CountingOwnedListener::default());
1214        transport
1215            .register_owned_listener(&UUri::any(), None, listener.clone())
1216            .await
1217            .expect("listener registered");
1218        let registered_listener = transport
1219            .listener
1220            .lock()
1221            .expect("listener lock poisoned")
1222            .clone()
1223            .expect("implementation should receive validating listener");
1224
1225        // Invalid-frame delivery is unrepresentable now: `on_receive_owned`
1226        // takes `UOwnedFrame<Validated>` by type.
1227        registered_listener.on_receive_owned(valid_frame()).await;
1228        assert_eq!(listener.count(), 1);
1229    }
1230
1231    #[tokio::test]
1232    async fn mock_owned_transport_delegates_send() {
1233        let frame = valid_frame();
1234        let expected = frame.clone();
1235        let mut transport = MockUOwnedTransport::new();
1236        transport
1237            .expect_do_send_validated_owned()
1238            .once()
1239            .withf(move |actual| actual == &expected)
1240            .return_const(Ok(()));
1241
1242        transport.send_owned(frame).await.unwrap();
1243    }
1244}
up_rust/utransport.rs

up_rust/
utransport.rs
