Use types for default implementations of L2 traits

This allows us to avoid dynamic dispatch and use static dispatch
instead, which can improve performance and reduce code size.

The InMemorySubscriber::for_clients function no longer takes a
LocalUriProvider parameter, because it was not used anyways.

Author: sophokles73

src/communication/default_notifier.rs

@@ -26,19 +26,19 @@ use super::{
 
 /// A [`Notifier`] that uses the uProtocol Transport Layer API to send and receive
 /// notifications to/from (other) uEntities.
-pub struct SimpleNotifier {
-    transport: Arc<dyn UTransport>,
-    uri_provider: Arc<dyn LocalUriProvider>,
+pub struct SimpleNotifier<T, P> {
+    transport: Arc<T>,
+    uri_provider: Arc<P>,
 }
 
-impl SimpleNotifier {
+impl<T: UTransport, P: LocalUriProvider> SimpleNotifier<T, P> {
     /// Creates a new Notifier for a given transport.
     ///
     /// # Arguments
     ///
     /// * `transport` - The uProtocol Transport Layer implementation to use for sending and receiving notification messages.
     /// * `uri_provider` - The helper for creating URIs that represent local resources.
-    pub fn new(transport: Arc<dyn UTransport>, uri_provider: Arc<dyn LocalUriProvider>) -> Self {
+    pub fn new(transport: Arc<T>, uri_provider: Arc<P>) -> Self {
         SimpleNotifier {
             transport,
             uri_provider,
@@ -47,7 +47,7 @@ impl SimpleNotifier {
 }
 
 #[async_trait]
-impl Notifier for SimpleNotifier {
+impl<T: UTransport, P: LocalUriProvider> Notifier for SimpleNotifier<T, P> {
     async fn notify(
         &self,
         resource_id: u16,
@@ -111,7 +111,7 @@ mod tests {
         StaticUriProvider, UCode, UPriority, UStatus, UUri, UUID,
     };
 
-    fn new_uri_provider() -> Arc<dyn LocalUriProvider> {
+    fn new_uri_provider() -> Arc<StaticUriProvider> {
         Arc::new(StaticUriProvider::new("", 0x0005, 0x02))
     }
 

src/communication/default_pubsub.rs

@@ -184,19 +184,19 @@ impl UListener for SubscriptionChangeListener {
 }
 
 /// A [`Publisher`] that uses the uProtocol Transport Layer API for publishing events to topics.
-pub struct SimplePublisher {
-    transport: Arc<dyn UTransport>,
-    uri_provider: Arc<dyn LocalUriProvider>,
+pub struct SimplePublisher<T, P> {
+    transport: Arc<T>,
+    uri_provider: Arc<P>,
 }
 
-impl SimplePublisher {
+impl<T: UTransport, P: LocalUriProvider> SimplePublisher<T, P> {
     /// Creates a new client.
     ///
     /// # Arguments
     ///
     /// * `transport` - The transport to use for sending messages.
     /// * `uri_provider` - The service to use for creating the event messages' _sink_ address.
-    pub fn new(transport: Arc<dyn UTransport>, uri_provider: Arc<dyn LocalUriProvider>) -> Self {
+    pub fn new(transport: Arc<T>, uri_provider: Arc<P>) -> Self {
         SimplePublisher {
             transport,
             uri_provider,
@@ -205,7 +205,7 @@ impl SimplePublisher {
 }
 
 #[async_trait]
-impl Publisher for SimplePublisher {
+impl<T: UTransport, P: LocalUriProvider> Publisher for SimplePublisher<T, P> {
     async fn publish(
         &self,
         resource_id: u16,
@@ -242,15 +242,16 @@ impl Publisher for SimplePublisher {
 /// about the new subscription and a (client provided) subscription change handler is registered with the
 /// listener. When a subscription change notification arrives from the USubscription service, the corresponding
 /// handler is being looked up and invoked.
-pub struct InMemorySubscriber {
-    transport: Arc<dyn UTransport>,
-    _uri_provider: Arc<dyn LocalUriProvider>,
-    usubscription: Arc<dyn USubscription>,
-    notifier: Arc<dyn Notifier>,
+pub struct InMemorySubscriber<T, S, N> {
+    transport: Arc<T>,
+    usubscription: Arc<S>,
+    notifier: Arc<N>,
     subscription_change_listener: Arc<SubscriptionChangeListener>,
 }
 
-impl InMemorySubscriber {
+impl<T: UTransport + 'static, P: LocalUriProvider + 'static>
+    InMemorySubscriber<T, RpcClientUSubscription, SimpleNotifier<T, P>>
+{
     /// Creates a new Subscriber for a given transport.
     ///
     /// The subscriber keeps track of subscription change handlers in memory only.
@@ -260,35 +261,32 @@ impl InMemorySubscriber {
     /// # Errors
     ///
     /// Returns an error if the Notifier cannot register a listener for notifications from the USubscription service.
-    pub async fn new(
-        transport: Arc<dyn UTransport>,
-        uri_provider: Arc<dyn LocalUriProvider>,
-    ) -> Result<Self, RegistrationError> {
+    pub async fn new(transport: Arc<T>, uri_provider: Arc<P>) -> Result<Self, RegistrationError> {
         let rpc_client = InMemoryRpcClient::new(transport.clone(), uri_provider.clone())
             .await
             .map(Arc::new)?;
         let usubscription_client = Arc::new(RpcClientUSubscription::new(rpc_client));
         let notifier = Arc::new(SimpleNotifier::new(transport.clone(), uri_provider.clone()));
-        Self::for_clients(transport, uri_provider, usubscription_client, notifier).await
+        Self::for_clients(transport, usubscription_client, notifier).await
     }
+}
 
+impl<T: UTransport, S: USubscription, N: Notifier> InMemorySubscriber<T, S, N> {
     /// Creates a new Subscriber for given clients.
     ///
     /// # Arguments
     ///
     /// * `transport` - The transport to use for registering the event listeners for subscribed topics.
-    /// * `uri-provider` - The service to use for creating topic addresses.
     /// * `usubscription` - The client to use for interacting with the (local) USubscription service.
     /// * `notifier` - The client to use for registering the listener for subscription updates from USubscription.
     ///
     /// # Errors
     ///
     /// Returns an error if the Notifier cannot register a listener for notifications from the USubscription service.
     pub async fn for_clients(
-        transport: Arc<dyn UTransport>,
-        uri_provider: Arc<dyn LocalUriProvider>,
-        usubscription: Arc<dyn USubscription>,
-        notifier: Arc<dyn Notifier>,
+        transport: Arc<T>,
+        usubscription: Arc<S>,
+        notifier: Arc<N>,
     ) -> Result<Self, RegistrationError> {
         // register a generic listener for subscription updates
         // whenever a uE later tries to subscribe to a topic, it can provide an optional callback for
@@ -304,7 +302,6 @@ impl InMemorySubscriber {
             .await?;
         Ok(InMemorySubscriber {
             transport,
-            _uri_provider: uri_provider,
             usubscription,
             notifier,
             subscription_change_listener,
@@ -400,7 +397,7 @@ impl InMemorySubscriber {
 }
 
 #[async_trait]
-impl Subscriber for InMemorySubscriber {
+impl<T: UTransport, U: USubscription, N: Notifier> Subscriber for InMemorySubscriber<T, U, N> {
     async fn subscribe(
         &self,
         topic_filter: &UUri,
@@ -459,11 +456,11 @@ mod tests {
         StaticUriProvider, UAttributes, UCode, UMessageType, UPriority, UStatus, UUri, UUID,
     };
 
-    fn new_uri_provider() -> Arc<dyn LocalUriProvider> {
+    fn new_uri_provider() -> Arc<StaticUriProvider> {
         Arc::new(StaticUriProvider::new("", 0x0005, 0x02))
     }
 
-    fn succeeding_notifier() -> Arc<dyn Notifier> {
+    fn succeeding_notifier() -> Arc<MockNotifier> {
         let mut notifier = MockNotifier::new();
         notifier
             .expect_start_listening()
@@ -586,7 +583,6 @@ mod tests {
         // WHEN trying to create a Subscriber for this Notifier
         let creation_attempt = InMemorySubscriber::for_clients(
             Arc::new(MockTransport::new()),
-            new_uri_provider(),
             Arc::new(MockUSubscription::new()),
             Arc::new(notifier),
         )
@@ -612,7 +608,6 @@ mod tests {
 
         let subscriber = InMemorySubscriber {
             transport: Arc::new(MockTransport::new()),
-            _uri_provider: new_uri_provider(),
             usubscription: Arc::new(MockUSubscription::new()),
             notifier: Arc::new(notifier),
             subscription_change_listener,
@@ -682,7 +677,6 @@ mod tests {
         // and a Subscriber using that USubscription client
         let subscriber = InMemorySubscriber::for_clients(
             Arc::new(transport),
-            new_uri_provider(),
             Arc::new(usubscription_client),
             succeeding_notifier(),
         )
@@ -752,7 +746,6 @@ mod tests {
         // and a Subscriber using that USubscription client, Notifier and transport
         let subscriber = InMemorySubscriber::for_clients(
             Arc::new(transport),
-            new_uri_provider(),
             Arc::new(usubscription_client),
             succeeding_notifier(),
         )
@@ -834,7 +827,6 @@ mod tests {
         // and a Subscriber using that USubscription client, Notifier and transport
         let subscriber = InMemorySubscriber::for_clients(
             Arc::new(transport),
-            new_uri_provider(),
             Arc::new(usubscription_client),
             succeeding_notifier(),
         )
@@ -891,7 +883,6 @@ mod tests {
         // and a Subscriber using that USubscription client, Notifier and transport
         let subscriber = InMemorySubscriber::for_clients(
             Arc::new(transport),
-            new_uri_provider(),
             Arc::new(usubscription_client),
             succeeding_notifier(),
         )
@@ -928,7 +919,6 @@ mod tests {
         // and a Subscriber using that USubscription client, Notifier and transport
         let subscriber = InMemorySubscriber::for_clients(
             Arc::new(transport),
-            new_uri_provider(),
             Arc::new(usubscription_client),
             succeeding_notifier(),
         )
@@ -975,7 +965,6 @@ mod tests {
         // and a Subscriber using that USubscription client, Notifier and transport
         let subscriber = InMemorySubscriber::for_clients(
             Arc::new(transport),
-            new_uri_provider(),
             Arc::new(usubscription_client),
             succeeding_notifier(),
         )
@@ -1031,7 +1020,6 @@ mod tests {
         // and a Subscriber using that USubscription client, Notifier and transport
         let subscriber = InMemorySubscriber::for_clients(
             Arc::new(transport),
-            new_uri_provider(),
             Arc::new(usubscription_client),
             succeeding_notifier(),
         )

src/communication/in_memory_rpc_client.rs

@@ -156,13 +156,13 @@ impl UListener for ResponseListener {
 /// implementation and a response handler is created and registered with the listener.
 /// When an RPC Response message arrives from the service, the corresponding handler is being looked
 /// up and invoked.
-pub struct InMemoryRpcClient {
-    transport: Arc<dyn UTransport>,
-    uri_provider: Arc<dyn LocalUriProvider>,
+pub struct InMemoryRpcClient<T, P> {
+    transport: Arc<T>,
+    uri_provider: Arc<P>,
     response_listener: Arc<ResponseListener>,
 }
 
-impl InMemoryRpcClient {
+impl<T: UTransport, P: LocalUriProvider> InMemoryRpcClient<T, P> {
     /// Creates a new RPC client for a given transport.
     ///
     /// # Arguments
@@ -174,10 +174,7 @@ impl InMemoryRpcClient {
     ///
     /// Returns an error if the generic RPC Response listener could not be
     /// registered with the given transport.
-    pub async fn new(
-        transport: Arc<dyn UTransport>,
-        uri_provider: Arc<dyn LocalUriProvider>,
-    ) -> Result<Self, RegistrationError> {
+    pub async fn new(transport: Arc<T>, uri_provider: Arc<P>) -> Result<Self, RegistrationError> {
         let response_listener = Arc::new(ResponseListener {
             pending_requests: Mutex::new(HashMap::new()),
         });
@@ -204,7 +201,7 @@ impl InMemoryRpcClient {
 }
 
 #[async_trait]
-impl RpcClient for InMemoryRpcClient {
+impl<T: UTransport, P: LocalUriProvider> RpcClient for InMemoryRpcClient<T, P> {
     async fn invoke_method(
         &self,
         method: UUri,
@@ -282,7 +279,7 @@ mod tests {
 
     use crate::{utransport::MockTransport, StaticUriProvider, UMessageBuilder, UPriority, UUri};
 
-    fn new_uri_provider() -> Arc<dyn LocalUriProvider> {
+    fn new_uri_provider() -> Arc<StaticUriProvider> {
         Arc::new(StaticUriProvider::new("", 0x0005, 0x02))
     }
 

src/communication/in_memory_rpc_server.rs

@@ -28,12 +28,12 @@ use crate::{
 
 use super::{RegistrationError, RequestHandler, RpcServer, ServiceInvocationError, UPayload};
 
-struct RequestListener {
+struct RequestListener<T: UTransport> {
     request_handler: Arc<dyn RequestHandler>,
-    transport: Arc<dyn UTransport>,
+    transport: Arc<T>,
 }
 
-impl RequestListener {
+impl<T: UTransport> RequestListener<T> {
     async fn process_valid_request(&self, resource_id: u16, request_message: UMessage) {
         let transport_clone = self.transport.clone();
         let request_handler_clone = self.request_handler.clone();
@@ -93,7 +93,7 @@ impl RequestListener {
 }
 
 #[async_trait]
-impl UListener for RequestListener {
+impl<T: UTransport> UListener for RequestListener<T> {
     async fn on_receive(&self, msg: UMessage) {
         if msg.is_request() {
             // cannot fail because inbound messages are validated at the transport layer already
@@ -117,15 +117,15 @@ impl UListener for RequestListener {
 /// the given request handler and registered with the underlying transport. The listener is also
 /// mapped to the endpoint's method resource ID in order to prevent registration of multiple
 /// request handlers for the same method.
-pub struct InMemoryRpcServer {
-    transport: Arc<dyn UTransport>,
-    uri_provider: Arc<dyn LocalUriProvider>,
+pub struct InMemoryRpcServer<T, P> {
+    transport: Arc<T>,
+    uri_provider: Arc<P>,
     request_listeners: tokio::sync::Mutex<HashMap<u16, Arc<dyn UListener>>>,
 }
 
-impl InMemoryRpcServer {
+impl<T: UTransport, P: LocalUriProvider> InMemoryRpcServer<T, P> {
     /// Creates a new RPC server for a given transport.
-    pub fn new(transport: Arc<dyn UTransport>, uri_provider: Arc<dyn LocalUriProvider>) -> Self {
+    pub fn new(transport: Arc<T>, uri_provider: Arc<P>) -> Self {
         InMemoryRpcServer {
             transport,
             uri_provider,
@@ -161,7 +161,7 @@ impl InMemoryRpcServer {
 }
 
 #[async_trait]
-impl RpcServer for InMemoryRpcServer {
+impl<T: UTransport + 'static, P: LocalUriProvider> RpcServer for InMemoryRpcServer<T, P> {
     async fn register_endpoint(
         &self,
         origin_filter: Option<&UUri>,
@@ -244,7 +244,7 @@ mod tests {
         UAttributes, UCode, UUri, UUID,
     };
 
-    fn new_uri_provider() -> Arc<dyn LocalUriProvider> {
+    fn new_uri_provider() -> Arc<StaticUriProvider> {
         Arc::new(StaticUriProvider::new("", 0x0005, 0x02))
     }