Skip to content

Use StaticUriProvider in examples #206

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

Merged
merged 1 commit into from
Oct 7, 2024
Merged

Use StaticUriProvider in examples #206

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
8 changes: 4 additions & 4 deletions examples/simple_notify.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ use protobuf::well_known_types::wrappers::StringValue;
use up_rust::{
communication::{CallOptions, Notifier, SimpleNotifier, UPayload},
local_transport::LocalTransport,
LocalUriProvider, UListener, UMessage,
LocalUriProvider, StaticUriProvider, UListener, UMessage,
};

struct ConsolePrinter {}
Expand All @@ -35,9 +35,9 @@ impl UListener for ConsolePrinter {
pub async fn main() -> Result<(), Box<dyn std::error::Error>> {
const ORIGIN_RESOURCE_ID: u16 = 0xd100;

let transport = Arc::new(LocalTransport::new("my-vehicle", 0xa34b, 0x01));
let uri_provider: Arc<dyn LocalUriProvider> = transport.clone();
let notifier = SimpleNotifier::new(transport.clone(), uri_provider.clone());
let uri_provider = Arc::new(StaticUriProvider::new("my-vehicle", 0xa34b, 0x01));
let transport = Arc::new(LocalTransport::default());
let notifier = SimpleNotifier::new(transport, uri_provider.clone());
let topic = uri_provider.get_resource_uri(ORIGIN_RESOURCE_ID);
let listener = Arc::new(ConsolePrinter {});

Expand Down
6 changes: 3 additions & 3 deletions examples/simple_publish.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ use protobuf::well_known_types::wrappers::StringValue;
use up_rust::{
communication::{CallOptions, Publisher, SimplePublisher, UPayload},
local_transport::LocalTransport,
LocalUriProvider, UListener, UMessage, UTransport,
LocalUriProvider, StaticUriProvider, UListener, UMessage, UTransport,
};

struct ConsolePrinter {}
Expand All @@ -34,8 +34,8 @@ impl UListener for ConsolePrinter {
#[tokio::main]
pub async fn main() -> Result<(), Box<dyn std::error::Error>> {
const ORIGIN_RESOURCE_ID: u16 = 0xb4c1;
let transport = Arc::new(LocalTransport::new("my-vehicle", 0xa34b, 0x01));
let uri_provider: Arc<dyn LocalUriProvider> = transport.clone();
let uri_provider = Arc::new(StaticUriProvider::new("my-vehicle", 0xa34b, 0x01));
let transport = Arc::new(LocalTransport::default());
let publisher = SimplePublisher::new(transport.clone(), uri_provider.clone());
let listener = Arc::new(ConsolePrinter {});

Expand Down
8 changes: 4 additions & 4 deletions examples/simple_rpc.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ use up_rust::{
ServiceInvocationError, UPayload,
},
local_transport::LocalTransport,
LocalUriProvider,
LocalUriProvider, StaticUriProvider,
};

struct EchoOperation {}
Expand All @@ -49,8 +49,8 @@ impl RequestHandler for EchoOperation {
#[tokio::main]
pub async fn main() -> Result<(), Box<dyn std::error::Error>> {
const METHOD_RESOURCE_ID: u16 = 0x00a0;
let transport = Arc::new(LocalTransport::new("my-vehicle", 0xa34b, 0x01));
let uri_provider: Arc<dyn LocalUriProvider> = transport.clone();
let uri_provider = Arc::new(StaticUriProvider::new("my-vehicle", 0xa34b, 0x01));
let transport = Arc::new(LocalTransport::default());

// create the RpcServer using the local transport
let rpc_server = InMemoryRpcServer::new(transport.clone(), uri_provider.clone());
Expand All @@ -62,7 +62,7 @@ pub async fn main() -> Result<(), Box<dyn std::error::Error>> {
.await?;

// now create an RpcClient attached to the same local transport
let rpc_client = InMemoryRpcClient::new(transport.clone(), uri_provider.clone()).await?;
let rpc_client = InMemoryRpcClient::new(transport, uri_provider.clone()).await?;
// and invoke the service operation without any payload
match rpc_client
.invoke_method(
Expand Down
61 changes: 19 additions & 42 deletions src/local_transport.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ use std::{collections::HashSet, sync::Arc};

use tokio::sync::RwLock;

use crate::{ComparableListener, LocalUriProvider, UListener, UMessage, UStatus, UTransport, UUri};
use crate::{ComparableListener, UListener, UMessage, UStatus, UTransport, UUri};

#[derive(Eq, PartialEq, Hash)]
struct RegisteredListener {
Expand Down Expand Up @@ -66,22 +66,12 @@ impl RegisteredListener {
///
/// A message sent via [`UTransport::send`] will be dispatched to all registered listeners that
/// match the message's source and sink filters.
#[derive(Default)]
pub struct LocalTransport {
listeners: RwLock<HashSet<RegisteredListener>>,
authority_name: String,
entity_id: u32,
entity_version: u8,
}

impl LocalTransport {
pub fn new(authority_name: &str, entity_id: u32, entity_version: u8) -> Self {
LocalTransport {
listeners: RwLock::new(HashSet::new()),
authority_name: authority_name.to_string(),
entity_id,
entity_version,
}
}
async fn dispatch(&self, message: UMessage) {
let listeners = self.listeners.read().await;
for listener in listeners.iter() {
Expand All @@ -92,25 +82,6 @@ impl LocalTransport {
}
}

impl LocalUriProvider for LocalTransport {
fn get_authority(&self) -> String {
self.authority_name.clone()
}

fn get_resource_uri(&self, resource_id: u16) -> UUri {
UUri::try_from_parts(
&self.authority_name,
self.entity_id,
self.entity_version,
resource_id,
)
.unwrap()
}
fn get_source_uri(&self) -> UUri {
self.get_resource_uri(0x0000)
}
}

#[async_trait::async_trait]
impl UTransport for LocalTransport {
async fn send(&self, message: UMessage) -> Result<(), UStatus> {
Expand Down Expand Up @@ -167,39 +138,44 @@ impl UTransport for LocalTransport {
#[cfg(test)]
mod tests {
use super::*;
use crate::{utransport::MockUListener, UMessageBuilder};
use crate::{utransport::MockUListener, LocalUriProvider, StaticUriProvider, UMessageBuilder};

#[tokio::test]
async fn test_send_dispatches_to_matching_listener() {
const RESOURCE_ID: u16 = 0xa1b3;
let mut listener = MockUListener::new();
listener.expect_on_receive().once().return_const(());
let listener_ref = Arc::new(listener);
let transport = LocalTransport::new("my-vehicle", 0x100d, 0x02);
let uri_provider = StaticUriProvider::new("my-vehicle", 0x100d, 0x02);
let transport = LocalTransport::default();

transport
.register_listener(
&transport.get_resource_uri(RESOURCE_ID),
&uri_provider.get_resource_uri(RESOURCE_ID),
None,
listener_ref.clone(),
)
.await
.unwrap();
let _ = transport
.send(
UMessageBuilder::publish(transport.get_resource_uri(RESOURCE_ID))
UMessageBuilder::publish(uri_provider.get_resource_uri(RESOURCE_ID))
.build()
.unwrap(),
)
.await;

transport
.unregister_listener(&transport.get_resource_uri(RESOURCE_ID), None, listener_ref)
.unregister_listener(
&uri_provider.get_resource_uri(RESOURCE_ID),
None,
listener_ref,
)
.await
.unwrap();
let _ = transport
.send(
UMessageBuilder::publish(transport.get_resource_uri(RESOURCE_ID))
UMessageBuilder::publish(uri_provider.get_resource_uri(RESOURCE_ID))
.build()
.unwrap(),
)
Expand All @@ -212,35 +188,36 @@ mod tests {
let mut listener = MockUListener::new();
listener.expect_on_receive().never().return_const(());
let listener_ref = Arc::new(listener);
let transport = LocalTransport::new("my-vehicle", 0x100d, 0x02);
let uri_provider = StaticUriProvider::new("my-vehicle", 0x100d, 0x02);
let transport = LocalTransport::default();

transport
.register_listener(
&transport.get_resource_uri(RESOURCE_ID + 10),
&uri_provider.get_resource_uri(RESOURCE_ID + 10),
None,
listener_ref.clone(),
)
.await
.unwrap();
let _ = transport
.send(
UMessageBuilder::publish(transport.get_resource_uri(RESOURCE_ID))
UMessageBuilder::publish(uri_provider.get_resource_uri(RESOURCE_ID))
.build()
.unwrap(),
)
.await;

transport
.unregister_listener(
&transport.get_resource_uri(RESOURCE_ID + 10),
&uri_provider.get_resource_uri(RESOURCE_ID + 10),
None,
listener_ref,
)
.await
.unwrap();
let _ = transport
.send(
UMessageBuilder::publish(transport.get_resource_uri(RESOURCE_ID))
UMessageBuilder::publish(uri_provider.get_resource_uri(RESOURCE_ID))
.build()
.unwrap(),
)
Expand Down