Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Update examples with new UListener interface and singleton UUIDBuilder. #4

Merged
merged 3 commits into from
May 2, 2024
Merged

Update examples with new UListener interface and singleton UUIDBuilder. #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0e6f348
Update examples with new UListener interface and singleton UUIDBuilder.
evshary Apr 17, 2024
535248b
Unify the UUri in examples.
evshary Apr 17, 2024
32b8b2c
Use the correct commit ID of up-client-zenoh
evshary Apr 17, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,10 +14,11 @@ pedantic = "deny"

[dependencies]
async-std = "1.12.0"
async-trait = "0.1"
chrono = "0.4.31"
env_logger = "0.10.0"
up-rust = { git = "https://github.com/eclipse-uprotocol/up-rust", rev = "68c8a1d94f0006daf4ba135c9cbbfddcd793108d" }
up-client-zenoh = { git = "https://github.com/eclipse-uprotocol/up-client-zenoh-rust", rev = "8855b9abd4bd27228d30d9061522194f330fa547" }
up-rust = { git = "https://github.com/eclipse-uprotocol/up-rust", rev = "c705ac97602ad6917a93d23651e8a504ec7bb718" }
up-client-zenoh = { git = "https://github.com/eclipse-uprotocol/up-client-zenoh-rust", rev = "dc0a3a95564ab1c83771fe71f0da5a261d469fb6" }
zenoh = { version = "0.10.1-rc", features = ["unstable"]}

[[bin]]
Expand Down
63 changes: 63 additions & 0 deletions src/common_uuri.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,63 @@
//
// Copyright (c) 2024 ZettaScale Technology
//
// This program and the accompanying materials are made available under the
// terms of the Eclipse Public License 2.0 which is available at
// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
// which is available at https://www.apache.org/licenses/LICENSE-2.0.
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
//
// Contributors:
// ZettaScale Zenoh Team, <[email protected]>
//

use up_rust::{Number, UAuthority, UEntity, UResource, UResourceBuilder};

pub enum ExampleType {
Publisher,
Subscriber,
RpcServer,
RpcClient,
}

#[allow(clippy::must_use_candidate)]
pub fn authority() -> UAuthority {
UAuthority {
name: Some("auth_name".to_string()),
number: Some(Number::Id(vec![1, 2, 3, 4])),
..Default::default()
}
}

#[allow(clippy::must_use_candidate)]
pub fn entity(example_type: &ExampleType) -> UEntity {
let (name, id) = match example_type {
ExampleType::Publisher => ("publisher", 1),
ExampleType::Subscriber => ("subscriber", 2),
ExampleType::RpcServer => ("rpc_server", 3),
ExampleType::RpcClient => ("rpc_client", 4),
};
UEntity {
name: name.to_string(),
id: Some(1),
version_major: Some(id),
..Default::default()
}
}

#[allow(clippy::must_use_candidate)]
pub fn pub_resource() -> UResource {
UResource {
name: "door".to_string(),
instance: Some("front_left".to_string()),
message: Some("Door".to_string()),
id: Some(5678),
..Default::default()
}
}

#[allow(clippy::must_use_candidate)]
pub fn rpc_resource() -> UResource {
UResourceBuilder::for_rpc_request(Some("getTime".to_string()), Some(5678))
}
38 changes: 15 additions & 23 deletions src/publisher.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,14 +11,13 @@
// Contributors:
// ZettaScale Zenoh Team, <[email protected]>
//
pub mod common_uuri;

use async_std::task;
use common_uuri::ExampleType;
use std::time;
use up_client_zenoh::UPClientZenoh;
use up_rust::{
transport::{builder::UMessageBuilder, datamodel::UTransport},
uprotocol::{UEntity, UPayloadFormat, UResource, UUri},
uuid::builder::UUIDBuilder,
};
use up_rust::{UMessageBuilder, UPayloadFormat, UTransport, UUIDBuilder, UUri};
use zenoh::config::Config;

#[async_std::main]
Expand All @@ -27,34 +26,27 @@ async fn main() {
env_logger::init();

println!("uProtocol publisher example");
let publisher = UPClientZenoh::new(Config::default()).await.unwrap();
let publisher = UPClientZenoh::new(
Config::default(),
common_uuri::authority(),
common_uuri::entity(&ExampleType::Publisher),
)
.await
.unwrap();

// create uuri
let uuri = UUri {
entity: Some(UEntity {
name: "body.access".to_string(),
version_major: Some(1),
id: Some(1234),
..Default::default()
})
.into(),
resource: Some(UResource {
name: "door".to_string(),
instance: Some("front_left".to_string()),
message: Some("Door".to_string()),
id: Some(5678),
..Default::default()
})
.into(),
entity: Some(common_uuri::entity(&ExampleType::Publisher)).into(),
resource: Some(common_uuri::pub_resource()).into(),
..Default::default()
};

let mut cnt: u64 = 0;
loop {
let data = format!("{cnt}");
let umessage = UMessageBuilder::publish(&uuri)
let umessage = UMessageBuilder::publish(uuri.clone())
.with_message_id(UUIDBuilder::build())
.build_with_payload(
&UUIDBuilder::new(),
data.as_bytes().to_vec().into(),
UPayloadFormat::UPAYLOAD_FORMAT_TEXT,
)
Expand Down
40 changes: 21 additions & 19 deletions src/rpc_client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,12 +11,11 @@
// Contributors:
// ZettaScale Zenoh Team, <[email protected]>
//
pub mod common_uuri;

use common_uuri::ExampleType;
use up_client_zenoh::UPClientZenoh;
use up_rust::{
rpc::{CallOptionsBuilder, RpcClient},
uprotocol::{Data, UEntity, UPayload, UPayloadFormat, UUri},
uri::builder::resourcebuilder::UResourceBuilder,
};
use up_rust::{CallOptions, Data, RpcClient, UPayload, UPayloadFormat, UUri};
use zenoh::config::Config;

#[async_std::main]
Expand All @@ -25,22 +24,18 @@ async fn main() {
env_logger::init();

println!("uProtocol RPC client example");
let rpc_client = UPClientZenoh::new(Config::default()).await.unwrap();
let rpc_client = UPClientZenoh::new(
Config::default(),
common_uuri::authority(),
common_uuri::entity(&ExampleType::RpcClient),
)
.await
.unwrap();

// create uuri
let uuri = UUri {
entity: Some(UEntity {
name: "test_rpc.app".to_string(),
version_major: Some(1),
id: Some(1234),
..Default::default()
})
.into(),
resource: Some(UResourceBuilder::for_rpc_request(
Some("getTime".to_string()),
Some(5678),
))
.into(),
entity: Some(common_uuri::entity(&ExampleType::RpcServer)).into(),
resource: Some(common_uuri::rpc_resource()).into(),
..Default::default()
};

Expand All @@ -56,7 +51,14 @@ async fn main() {
// invoke RPC method
println!("Send request to {uuri}");
let result = rpc_client
.invoke_method(uuri, payload, CallOptionsBuilder::default().build())
.invoke_method(
uuri,
payload,
CallOptions {
ttl: 1000,
..Default::default()
},
)
.await;

// process the result
Expand Down
118 changes: 57 additions & 61 deletions src/rpc_server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,89 +11,85 @@
// Contributors:
// ZettaScale Zenoh Team, <[email protected]>
//
pub mod common_uuri;

use async_std::task::{self, block_on};
use async_trait::async_trait;
use chrono::Utc;
use common_uuri::ExampleType;
use std::{sync::Arc, time};
use up_client_zenoh::UPClientZenoh;
use up_rust::{
rpc::RpcServer,
transport::datamodel::UTransport,
uprotocol::{Data, UEntity, UMessage, UMessageType, UPayload, UPayloadFormat, UStatus, UUri},
uri::builder::resourcebuilder::UResourceBuilder,
Data, UListener, UMessage, UMessageBuilder, UPayloadFormat, UStatus, UTransport, UUIDBuilder,
UUri,
};
use zenoh::config::Config;

struct RpcListener {
up_client: Arc<UPClientZenoh>,
}
impl RpcListener {
fn new(up_client: Arc<UPClientZenoh>) -> Self {
RpcListener { up_client }
}
}
#[async_trait]
impl UListener for RpcListener {
async fn on_receive(&self, msg: UMessage) {
let UMessage {
attributes,
payload,
..
} = msg;
// Build the payload to send back
if let Data::Value(v) = payload.unwrap().data.unwrap() {
let value = v.into_iter().map(|c| c as char).collect::<String>();
let source = attributes.clone().unwrap().source.unwrap();
let sink = attributes.clone().unwrap().sink.unwrap();
println!("Receive {value} from {source} to {sink}");
}
// Send back result
let umessage = UMessageBuilder::response_for_request(&attributes)
.with_message_id(UUIDBuilder::build())
.build_with_payload(
// Get current time
format!("{}", Utc::now()).as_bytes().to_vec().into(),
UPayloadFormat::UPAYLOAD_FORMAT_TEXT,
)
.unwrap();
block_on(self.up_client.send(umessage)).unwrap();
}
async fn on_error(&self, err: UStatus) {
panic!("Internal Error: {err:?}");
}
}

#[async_std::main]
async fn main() {
// initiate logging
env_logger::init();

println!("uProtocol RPC server example");
let rpc_server = Arc::new(UPClientZenoh::new(Config::default()).await.unwrap());
let rpc_server = Arc::new(
UPClientZenoh::new(
Config::default(),
common_uuri::authority(),
common_uuri::entity(&ExampleType::RpcServer),
)
.await
.unwrap(),
);

// create uuri
let uuri = UUri {
entity: Some(UEntity {
name: "test_rpc.app".to_string(),
version_major: Some(1),
id: Some(1234),
..Default::default()
})
.into(),
resource: Some(UResourceBuilder::for_rpc_request(
Some("getTime".to_string()),
Some(5678),
))
.into(),
entity: Some(common_uuri::entity(&ExampleType::RpcServer)).into(),
resource: Some(common_uuri::rpc_resource()).into(),
..Default::default()
};

let rpc_server_cloned = rpc_server.clone();
let callback = move |result: Result<UMessage, UStatus>| {
match result {
Ok(msg) => {
let UMessage {
attributes,
payload,
..
} = msg;
// Get the UUri
let source = attributes.clone().unwrap().source.unwrap();
let sink = attributes.clone().unwrap().sink.unwrap();
// Build the payload to send back
if let Data::Value(v) = payload.unwrap().data.unwrap() {
let value = v.into_iter().map(|c| c as char).collect::<String>();
println!("Receive {value} from {source} to {sink}");
}
// Get current time
let upayload = UPayload {
length: Some(0),
format: UPayloadFormat::UPAYLOAD_FORMAT_TEXT.into(),
data: Some(Data::Value(format!("{}", Utc::now()).as_bytes().to_vec())),
..Default::default()
};
// Set the attributes type to Response
let mut uattributes = attributes.unwrap();
uattributes.type_ = UMessageType::UMESSAGE_TYPE_RESPONSE.into();
uattributes.sink = Some(source.clone()).into();
uattributes.source = Some(sink.clone()).into();
// Send back result
block_on(rpc_server_cloned.send(UMessage {
attributes: Some(uattributes).into(),
payload: Some(upayload).into(),
..Default::default()
}))
.unwrap();
}
Err(ustatus) => {
println!("Internal Error: {ustatus:?}");
}
}
};

println!("Register the listener...");
rpc_server
.register_rpc_listener(uuri, Box::new(callback))
.register_listener(uuri, Arc::new(RpcListener::new(rpc_server.clone())))
.await
.unwrap();

Expand Down
Loading