Do not implement LocalUriProvider (#123)

* [#122] Do not implement LocalUriProvider

UPZenohTransport has been changed to no longer implement the
LocalUriProvider trait. This seems reasonable to allow for the
same transport instance to be used by multiple service implementations
having different entity types.

* Incorporate feedback

Author: sophokles73

Cargo.lock

@@ -46,6 +46,8 @@ chrono = "0.4.41"
 clap = { version = "4.5.40", features = ["derive"] }
 serde_json = "1.0.128"
 test-case = { version = "3.3" }
+tokio = { version = "1.45.1", default-features = false, features = ["signal"] }
+tracing-subscriber = "0.3.19"
 up-rust = { version = "0.6.0", features = ["communication", "test-util"] }
 
 [features]

Cargo.toml

@@ -24,7 +24,7 @@ mod common;
 use std::{str::FromStr, sync::Arc};
 use up_rust::{
     communication::{CallOptions, InMemoryRpcClient, RpcClient, UPayload},
-    LocalUriProvider, UPayloadFormat, UPriority, UUri, UUID,
+    LocalUriProvider, StaticUriProvider, UPayloadFormat, UPriority, UUri, UUID,
 };
 use up_transport_zenoh::UPTransportZenoh;
 
@@ -34,46 +34,47 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     UPTransportZenoh::try_init_log_from_env();
 
     println!("uProtocol RPC client example");
-    let zenoh_transport = UPTransportZenoh::builder("//rpc_client/1/1/0")
-        .expect("invalid URI")
+    let uri_provider = Arc::new(StaticUriProvider::new("l2_rpc_client", 0x10_ab10, 1));
+    let transport = UPTransportZenoh::builder(uri_provider.get_authority())
+        .expect("invalid authority name")
         .with_config(common::get_zenoh_config())
         .build()
         .await
         .map(Arc::new)?;
-
-    let rpc_client = InMemoryRpcClient::new(zenoh_transport.clone(), zenoh_transport.clone())
+    let rpc_client = InMemoryRpcClient::new(transport, uri_provider.clone())
         .await
         .map(Arc::new)?;
 
-    let sink_uuri = UUri::from_str("//rpc_server/1/1/1")?;
+    let operation_uuri = UUri::from_str("//rpc_server/AAA/1/6A10")?;
 
-    // create uPayload and send request
-    let data = String::from("GetCurrentTime");
-    let payload = UPayload::new(data, UPayloadFormat::UPAYLOAD_FORMAT_TEXT);
+    // create and send request
+    let payload = UPayload::new("GetCurrentTime", UPayloadFormat::UPAYLOAD_FORMAT_TEXT);
     let call_options = CallOptions::for_rpc_request(
         5_000,
         Some(UUID::build()),
         Some("my_token".to_string()),
         Some(UPriority::UPRIORITY_CS6),
     );
     println!(
-        "Sending request from {} to {}",
-        zenoh_transport.get_source_uri(),
-        sink_uuri
+        "Sending request [source: {}, sink: {}]",
+        uri_provider.get_source_uri().to_uri(false),
+        operation_uuri.to_uri(false)
     );
+
     match rpc_client
-        .invoke_method(sink_uuri, call_options, Some(payload))
+        .invoke_method(operation_uuri, call_options, Some(payload))
         .await
     {
-        Ok(result) => {
-            let payload = result.unwrap().payload();
-            let value = payload.into_iter().map(|c| c as char).collect::<String>();
-            println!("Receive {value}");
-            Ok(())
+        Err(_) => {
+            println!("Failed to receive reply from service");
+        }
+        Ok(Some(payload)) => {
+            let value = String::from_utf8(payload.payload().to_vec())?;
+            println!("Received reply [payload: {value}]");
         }
-        Err(e) => {
-            println!("Failed to receive the reply");
-            Err(Box::from(e))
+        _ => {
+            println!("Reply did not contain payload");
         }
     }
+    Ok(())
 }

examples/l2_rpc_client.rs

@@ -10,49 +10,61 @@
  *
  * SPDX-License-Identifier: Apache-2.0
  ********************************************************************************/
+
+/*!
+This example illustrates how uProtocol's Transport Layer API can be used to receive
+notifications sent by another uEntity using the Zenoh transport.
+
+This example works in conjunction with the `notifier`, which should be started in
+another terminal after having started this receiver.
+*/
+
 mod common;
 
 use async_trait::async_trait;
 use std::{str::FromStr, sync::Arc};
-use tokio::time::{sleep, Duration};
-use up_rust::{LocalUriProvider, UListener, UMessage, UTransport, UUri};
+use up_rust::{LocalUriProvider, StaticUriProvider, UListener, UMessage, UTransport, UUri};
 use up_transport_zenoh::UPTransportZenoh;
 
 struct SubscriberListener;
 #[async_trait]
 impl UListener for SubscriberListener {
     async fn on_receive(&self, msg: UMessage) {
         let payload = msg.payload.unwrap();
-        let value = payload.into_iter().map(|c| c as char).collect::<String>();
-        let uri = msg.attributes.unwrap().source.unwrap().to_string();
-        println!("Receiving notification '{value}' from {uri}");
+        let value = String::from_utf8(payload.to_vec()).unwrap();
+        let uri = msg.attributes.unwrap().source.unwrap().to_uri(false);
+        println!("Received notification [source: {uri}, payload: {value}]");
     }
 }
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // initiate logging
     UPTransportZenoh::try_init_log_from_env();
+
     println!("uProtocol notification receiver example");
-    let receiver = UPTransportZenoh::builder("//receiver/2/1/0")
-        .expect("invalid URI")
+    let uri_provider = StaticUriProvider::new("receiver", 0x10_ab10, 1);
+    let transport = UPTransportZenoh::builder(uri_provider.get_authority())
+        .expect("invalid authority name")
         .with_config(common::get_zenoh_config())
         .build()
         .await?;
 
-    // create uuri
-    let source_uuri = UUri::from_str("//notification/1/1/8001")?;
+    let source_filter = UUri::from_str("//*/FFFFA1B2/1/8001")?;
+    let sink_filter = uri_provider.get_source_uri();
 
-    println!("Register the listener...");
-    receiver
+    println!(
+        "Registering notification listener [source filter: {}, sink filter: {}]",
+        source_filter.to_uri(false),
+        sink_filter.to_uri(false)
+    );
+    transport
         .register_listener(
-            &source_uuri,
-            Some(&receiver.get_source_uri()),
+            &source_filter,
+            Some(&sink_filter),
             Arc::new(SubscriberListener {}),
         )
         .await?;
 
-    loop {
-        sleep(Duration::from_millis(1000)).await;
-    }
+    tokio::signal::ctrl_c().await.map_err(Box::from)
 }

examples/notification_receiver.rs

@@ -10,11 +10,22 @@
  *
  * SPDX-License-Identifier: Apache-2.0
  ********************************************************************************/
+
+/*!
+This example illustrates how uProtocol's Transport Layer API can be used to send
+notifications to another uEntity using the Zenoh transport.
+
+This example works in conjunction with the `notification_receiver`, which should
+be started in another terminal first.
+*/
+
 mod common;
 
 use std::str::FromStr;
 
-use up_rust::{LocalUriProvider, UMessageBuilder, UPayloadFormat, UTransport, UUri};
+use up_rust::{
+    LocalUriProvider, StaticUriProvider, UMessageBuilder, UPayloadFormat, UTransport, UUri,
+};
 use up_transport_zenoh::UPTransportZenoh;
 
 #[tokio::main]
@@ -23,19 +34,28 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     UPTransportZenoh::try_init_log_from_env();
 
     println!("uProtocol notifier example");
-    let notifier = UPTransportZenoh::builder("//notification/1/1/0")
-        .expect("invalid URI")
+    let uri_provider = StaticUriProvider::new("notification", 0xa1b2, 1);
+    let transport = UPTransportZenoh::builder(uri_provider.get_authority())
+        .expect("invalid authority name")
         .with_config(common::get_zenoh_config())
         .build()
         .await?;
 
     // create uuri
-    let sink_uuri = UUri::from_str("//receiver/2/1/0")?;
+    let source_uuri = uri_provider.get_resource_uri(0x8001);
+    let sink_uuri = UUri::from_str("//receiver/10AB10/1/0")?;
 
-    let data = "The notification data";
-    let umessage =
-        UMessageBuilder::notification(notifier.get_resource_uri(0x8001), sink_uuri.clone())
+    for cnt in 1..=100 {
+        let data = format!("notification {cnt}");
+        println!(
+            "Sending notification [from: {}, to: {}, payload: {data}]",
+            &source_uuri.to_uri(false),
+            &sink_uuri.to_uri(false)
+        );
+        let umessage = UMessageBuilder::notification(source_uuri.clone(), sink_uuri.clone())
             .build_with_payload(data, UPayloadFormat::UPAYLOAD_FORMAT_TEXT)?;
-    println!("Sending notification '{data}' to {sink_uuri}...");
-    notifier.send(umessage).await.map_err(Box::from)
+        transport.send(umessage).await?;
+        tokio::time::sleep(core::time::Duration::from_secs(1)).await;
+    }
+    Ok(())
 }

examples/notifier.rs

@@ -10,10 +10,18 @@
  *
  * SPDX-License-Identifier: Apache-2.0
  ********************************************************************************/
+
+/*!
+This example illustrates how uProtocol's Transport Layer API can be used to publish
+messages to a topic using the Zenoh transport.
+
+This example works in conjunction with the `subscriber`, which should be started in
+another terminal first.
+*/
+
 mod common;
 
-use tokio::time::{sleep, Duration};
-use up_rust::{LocalUriProvider, UMessageBuilder, UPayloadFormat, UTransport};
+use up_rust::{LocalUriProvider, StaticUriProvider, UMessageBuilder, UPayloadFormat, UTransport};
 use up_transport_zenoh::UPTransportZenoh;
 
 #[tokio::main]
@@ -22,24 +30,26 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     UPTransportZenoh::try_init_log_from_env();
 
     println!("uProtocol publisher example");
-    let publisher = UPTransportZenoh::builder("//publisher/1/1/0")
-        .expect("invalid URI")
+    let uri_provider = StaticUriProvider::new("publisher", 0x3_b1da, 1);
+    let transport = UPTransportZenoh::builder(uri_provider.get_authority())
+        .expect("invalid authority name")
         .with_config(common::get_zenoh_config())
         .build()
         .await?;
 
-    // create uuri
-    let uuri = publisher.get_resource_uri(0x8001);
-
-    let mut cnt: u64 = 0;
-    loop {
-        let data = format!("{cnt}");
-        let umessage = UMessageBuilder::publish(uuri.clone())
-            .build_with_payload(data.clone(), UPayloadFormat::UPAYLOAD_FORMAT_TEXT)
-            .unwrap();
-        println!("Publishing {data} from {uuri}...");
-        publisher.send(umessage).await.unwrap();
-        sleep(Duration::from_millis(1000)).await;
-        cnt += 1;
+    // create topic uuri
+    let topic = uri_provider.get_resource_uri(0x8001);
+
+    for cnt in 1..=100 {
+        let data = format!("event {cnt}");
+        println!(
+            "Publishing message [topic: {}, payload: {data}]",
+            topic.to_uri(false)
+        );
+        let umessage = UMessageBuilder::publish(topic.clone())
+            .build_with_payload(data, UPayloadFormat::UPAYLOAD_FORMAT_TEXT)?;
+        transport.send(umessage).await?;
+        tokio::time::sleep(core::time::Duration::from_secs(1)).await;
     }
+    Ok(())
 }