fixed tokio runtime issue and added publisher example

Xerner · Xerner · commit a8492e55435a · 2025-09-22T02:03:10.000Z
added publisher example
!!! now there is a segmentation fault when the subscriber receives a published message. yay, a new error
diff --git a/.vscode/launch.json b/.vscode/launch.json
@@ -37,6 +37,24 @@
             },
             "args": [],
             "cwd": "${workspaceFolder}"
+        },
+        {
+            "type": "lldb",
+            "request": "launch",
+            "name": "Debug example 'publisher'",
+            "cargo": {
+                "args": [
+                    "build",
+                    "--example=publisher",
+                    "--package=up-transport-iceoryx2-rust"
+                ],
+                "filter": {
+                    "name": "publisher",
+                    "kind": "example"
+                }
+            },
+            "args": [],
+            "cwd": "${workspaceFolder}"
         }
     ]
 }
diff --git a/examples/common/constants.rs b/examples/common/constants.rs
@@ -0,0 +1,21 @@
+// ################################################################################
+// Copyright (c) 2025 Contributors to the Eclipse Foundation
+//
+// See the NOTICE file(s) distributed with this work for additional
+// information regarding copyright ownership.
+//
+// This program and the accompanying materials are made available under the
+// terms of the Apache License Version 2.0 which is available at
+// https: //www.apache.org/licenses/LICENSE-2.0
+//
+// SPDX-License-Identifier: Apache-2.0
+// ################################################################################
+
+use std::time::Duration;
+
+/// The source filter publisher examples will use
+pub static SOURCE_FILTER_STR: &str = "up://device1/10AB/3/80CD";
+
+/// A UMessage will be published at this frequency
+#[allow(dead_code)]
+pub static CYCLE_TIME: Duration = Duration::from_secs(1);
diff --git a/examples/common/helpers.rs b/examples/common/helpers.rs
@@ -14,6 +14,7 @@
 use up_rust::UMessage;
 
 /// Simply prints the [`UMessage`] instances source uri, sink uri, and payload to STDOUT
+#[allow(dead_code)]
 pub fn print_umessage(msg: &UMessage) {
     let payload_utf8 = msg.payload.as_ref().map(|p| String::from_utf8_lossy(p));
     let (source_uri, sink_uri) = get_source_and_sink_uri(msg);
@@ -23,7 +24,7 @@ pub fn print_umessage(msg: &UMessage) {
     println!("Payload: {payload_utf8:?}");
 }
 
-pub fn get_source_and_sink_uri(msg: &UMessage) -> (Option<String>, Option<String>) {
+fn get_source_and_sink_uri(msg: &UMessage) -> (Option<String>, Option<String>) {
     let source_uri = msg
         .attributes
         .as_ref()
diff --git a/examples/common/mod.rs b/examples/common/mod.rs
@@ -1 +1,20 @@
+// ################################################################################
+// Copyright (c) 2025 Contributors to the Eclipse Foundation
+//
+// See the NOTICE file(s) distributed with this work for additional
+// information regarding copyright ownership.
+//
+// This program and the accompanying materials are made available under the
+// terms of the Apache License Version 2.0 which is available at
+// https: //www.apache.org/licenses/LICENSE-2.0
+//
+// SPDX-License-Identifier: Apache-2.0
+// ################################################################################
+
+pub mod constants;
 pub mod helpers;
+
+#[allow(unused_imports)]
+pub use constants::*;
+#[allow(unused_imports)]
+pub use helpers::*;
diff --git a/examples/publisher.rs b/examples/publisher.rs
@@ -0,0 +1,48 @@
+// ################################################################################
+// Copyright (c) 2025 Contributors to the Eclipse Foundation
+//
+// See the NOTICE file(s) distributed with this work for additional
+// information regarding copyright ownership.
+//
+// This program and the accompanying materials are made available under the
+// terms of the Apache License Version 2.0 which is available at
+// https: //www.apache.org/licenses/LICENSE-2.0
+//
+// SPDX-License-Identifier: Apache-2.0
+// ################################################################################
+
+use std::{error::Error, str::FromStr};
+use up_rust::{UMessage, UMessageBuilder, UPayloadFormat, UTransport, UUri};
+use up_transport_iceoryx2_rust::{MessagingPattern, transport::UTransportIceoryx2};
+
+mod common;
+use crate::common::*;
+
+fn create_umessage(source_filter: &UUri, counter: u64) -> Result<UMessage, Box<dyn Error>> {
+    let payload = format!("Hello, Iceoryx2! Message {counter}");
+    let umessage = UMessageBuilder::publish(source_filter.clone())
+        .build_with_payload(payload.into_bytes(), UPayloadFormat::UPAYLOAD_FORMAT_TEXT)?;
+    Ok(umessage)
+}
+
+#[tokio::main(flavor = "multi_thread")]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    println!("uProtocol UTransportIceoryx2 publisher example");
+    let source_filter = UUri::from_str(SOURCE_FILTER_STR).expect("Failed to create source UUri");
+    let transport = UTransportIceoryx2::build(MessagingPattern::PublishSubscribe)?;
+    let mut counter: u64 = 0;
+    let cycle_time_str = format!("{CYCLE_TIME:?}");
+    println!(
+        "Publishing a UMessage with an incrementing counter every '{cycle_time_str}' second with source filter '{SOURCE_FILTER_STR}'"
+    );
+    loop {
+        counter += 1;
+        let umessage = create_umessage(&source_filter, counter)?;
+        let memory_address = &umessage;
+        println!(
+            "Publishing message with source filter '{SOURCE_FILTER_STR}' and memory address {memory_address:p}"
+        );
+        transport.send(umessage).await?;
+        tokio::time::sleep(CYCLE_TIME).await;
+    }
+}
diff --git a/examples/subscriber.rs b/examples/subscriber.rs
@@ -13,36 +13,34 @@
 
 use async_trait::async_trait;
 use std::{str::FromStr, sync::Arc};
-use tracing::info;
 use up_rust::{UListener, UMessage, UTransport, UUri};
 use up_transport_iceoryx2_rust::{MessagingPattern, transport::UTransportIceoryx2};
 
-use crate::common::helpers::*;
-
 mod common;
+use crate::common::*;
 
-struct SubscriberListener(tokio::runtime::Runtime);
+struct SubscriberListener;
 
 #[async_trait]
 impl UListener for SubscriberListener {
     /// Spawns a task to process the received message. In this example, we simply print the message contents.
     async fn on_receive(&self, msg: UMessage) {
-        self.0.spawn(async move {
-            print_umessage(&msg);
-        });
+        print_umessage(&msg);
     }
 }
 
 #[tokio::main(flavor = "multi_thread")]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    info!("uProtocols UTransportIceoryx2 subscriber example");
-    let source_filter =
-        UUri::from_str("up://device1/10AB/3/80CD").expect("Failed to create source UUri");
+    println!("uProtocol UTransportIceoryx2 subscriber example");
+    let source_filter = UUri::from_str(SOURCE_FILTER_STR).expect("Failed to create source UUri");
     let transport = UTransportIceoryx2::build(MessagingPattern::PublishSubscribe)?;
-    let ulistener = Arc::new(SubscriberListener(tokio::runtime::Runtime::new()?));
+    let ulistener = Arc::new(SubscriberListener);
     transport
         .register_listener(&source_filter, None, ulistener)
         .await?;
-    info!("Listener registered. Waiting for messages...");
+    println!(
+        "Listening to message from source filter '{SOURCE_FILTER_STR}'. Press CTRL+C to kill this subscriber"
+    );
+    println!("Waiting for messages...");
     tokio::signal::ctrl_c().await.map_err(Box::from)
 }
diff --git a/src/service_name_mapping.rs b/src/service_name_mapping.rs
@@ -55,7 +55,7 @@ fn determine_message_type(
 }
 
 fn is_a_publish(source: &UUri, messaging_pattern: MessagingPattern) -> bool {
-    source.is_empty() == false && messaging_pattern == MessagingPattern::PublishSubscribe
+    !source.is_empty() && messaging_pattern == MessagingPattern::PublishSubscribe
 }
 
 pub fn compute_service_name(
diff --git a/src/transport.rs b/src/transport.rs
@@ -31,7 +31,7 @@ impl UTransportIceoryx2 {
         }
     }
 
-    fn build_publish_subscribe() -> Arc<impl UTransport> {
+    fn build_publish_subscribe() -> Arc<Iceoryx2PubSub> {
         Iceoryx2PubSub::new()
     }
 }
diff --git a/src/utransport_pubsub.rs b/src/utransport_pubsub.rs
@@ -32,7 +32,6 @@ use crate::{
 
 pub struct Iceoryx2PubSub {
     node: Node<ipc_threadsafe::Service>,
-    listener_worker_runtime: tokio::runtime::Runtime,
     pub publishers: PublisherSet<ipc_threadsafe::Service>,
     pub subscribers: SubscriberSet<ipc_threadsafe::Service>,
     pub listeners: ListenerMap,
@@ -43,28 +42,22 @@ impl Iceoryx2PubSub {
         let node = NodeBuilder::new()
             .create::<ipc_threadsafe::Service>()
             .expect("Failed to create Iceoryx2 Node");
-        let listener_worker_runtime =
-            tokio::runtime::Runtime::new().expect("Failed to create Tokio runtime");
         let transport = Arc::new(Self {
             node,
             publishers: RwLock::new(HashMap::new()),
             subscribers: RwLock::new(HashMap::new()),
             listeners: RwLock::new(HashMap::new()),
-            listener_worker_runtime,
         });
-        Iceoryx2WorkerDispatcher::start_listener_worker(
-            &transport.listener_worker_runtime,
-            transport.clone(),
-        );
+        Iceoryx2WorkerDispatcher::start_listener_worker(transport.clone());
         transport
     }
 
     pub fn create_subscriber(
         &self,
+        service_name: ServiceName,
     ) -> Result<Subscriber<ipc_threadsafe::Service, UMessageZeroCopy, UProtocolHeader>, UStatus>
     {
         // Placeholder implementation
-        let service_name: ServiceName = "example_service".try_into().unwrap();
         let service = self
             .node
             .service_builder(&service_name)
@@ -85,15 +78,21 @@ impl Iceoryx2PubSub {
         service_name: ServiceName,
     ) -> Result<Arc<Publisher<ipc_threadsafe::Service, UMessageZeroCopy, UProtocolHeader>>, UStatus>
     {
-        let publishers = self.publishers.read().await;
-        if publishers.contains_key(&service_name) {
-            let publisher = publishers.get(&service_name).unwrap();
-            return Ok(publisher.clone());
+        let publisher = self.get_publisher(service_name.clone()).await;
+        if let Some(publisher) = publisher {
+            return Ok(publisher);
         }
-        let service_name_res: Result<ServiceName, _> = service_name.as_str().try_into();
+        self.create_publisher(service_name).await
+    }
+
+    async fn create_publisher(
+        &self,
+        service_name: ServiceName,
+    ) -> Result<Arc<Publisher<ipc_threadsafe::Service, UMessageZeroCopy, UProtocolHeader>>, UStatus>
+    {
         let service = self
             .node
-            .service_builder(&service_name_res.unwrap())
+            .service_builder(&service_name)
             .publish_subscribe::<UMessageZeroCopy>()
             .user_header::<UProtocolHeader>()
             .open_or_create()
@@ -110,25 +109,30 @@ impl Iceoryx2PubSub {
         Ok(publisher.clone())
     }
 
+    async fn get_publisher(
+        &self,
+        service_name: ServiceName,
+    ) -> Option<Arc<Publisher<ipc_threadsafe::Service, UMessageZeroCopy, UProtocolHeader>>> {
+        let publishers = self.publishers.read().await;
+        if publishers.contains_key(&service_name) {
+            let publisher = publishers.get(&service_name).unwrap();
+            return Some(publisher.clone());
+        }
+        None
+    }
+
     pub async fn relay(&self) -> Result<(), UStatus> {
-        let current_thread_runtime = tokio::runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .map_err(|e| {
-                UStatus::fail_with_code(
-                    UCode::INTERNAL,
-                    format!("Failed to build current_thread runtime: {e}"),
-                )
-            })?;
-        for (service_name, subscriber) in self.subscribers.read().await.iter() {
+        let subscribers = self.subscribers.read().await;
+        for (service_name, subscriber) in subscribers.iter() {
             match subscriber.receive() {
                 Ok(Some(sample)) => {
                     let payload = sample;
                     if let Some(listeners_to_notify) = self.listeners.read().await.get(service_name)
                     {
                         for listener in listeners_to_notify.iter() {
                             let listener: &ComparableListener = listener;
-                            current_thread_runtime.block_on(listener.on_receive(payload.0.clone()));
+                            let payload_clone = payload.0.clone();
+                            listener.on_receive(payload_clone).await;
                         }
                     }
                 }
@@ -182,14 +186,17 @@ impl UTransport for Iceoryx2PubSub {
     ) -> Result<(), UStatus> {
         up_rust::verify_filter_criteria(source_filter, sink_filter)?;
         let service_name = compute_service_name(
-            &source_filter,
+            source_filter,
             sink_filter,
             MessagingPattern::PublishSubscribe,
         )?;
-        let subscribers = self.subscribers.read().await;
+        let has_subscriber = {
+            let subscribers = self.subscribers.read().await;
+            subscribers.contains_key(&service_name)
+        };
         // insert subscriber for service name if it does not already exist
-        if !subscribers.contains_key(&service_name) {
-            let subscriber = self.create_subscriber()?;
+        if !has_subscriber {
+            let subscriber = self.create_subscriber(service_name.clone())?;
             let mut subscribers = self.subscribers.write().await;
             subscribers.insert(service_name.clone(), Arc::new(subscriber));
         }
@@ -212,7 +219,7 @@ impl UTransport for Iceoryx2PubSub {
     ) -> Result<(), UStatus> {
         up_rust::verify_filter_criteria(source_filter, sink_filter)?;
         let service_name = compute_service_name(
-            &source_filter,
+            source_filter,
             sink_filter,
             MessagingPattern::PublishSubscribe,
         )?;
diff --git a/src/workers/dispatcher.rs b/src/workers/dispatcher.rs
@@ -12,18 +12,16 @@
 // ################################################################################
 
 use std::sync::{Arc, atomic::Ordering};
-use tokio::runtime::Runtime;
 use up_rust::UStatus;
 
 use crate::{utransport_pubsub::Iceoryx2PubSub, workers::worker::Iceoryx2Worker};
 
 pub struct Iceoryx2WorkerDispatcher {}
 
 impl Iceoryx2WorkerDispatcher {
-    pub fn start_listener_worker(runtime: &Runtime, transport: Arc<Iceoryx2PubSub>) {
+    pub fn start_listener_worker(transport: Arc<Iceoryx2PubSub>) {
         let worker = Iceoryx2Worker::new(transport.clone());
-        let future = Iceoryx2WorkerDispatcher::run(worker);
-        runtime.spawn(future);
+        tokio::spawn(async { Iceoryx2WorkerDispatcher::run(worker).await });
     }
 
     async fn run(worker: Iceoryx2Worker) -> Result<(), UStatus> {

Original file line number	Diff line number	Diff line change
`@@ -55,7 +55,7 @@ fn determine_message_type(`
`55`	`55`	`}`
`56`	`56`
`57`	`57`	`fn is_a_publish(source: &UUri, messaging_pattern: MessagingPattern) -> bool {`
`58`		`- source.is_empty() == false && messaging_pattern == MessagingPattern::PublishSubscribe`
	`58`	`+ !source.is_empty() && messaging_pattern == MessagingPattern::PublishSubscribe`
`59`	`59`	`}`
`60`	`60`
`61`	`61`	`pub fn compute_service_name(`
Original file line number	Diff line number	Diff line change
`@@ -31,7 +31,7 @@ impl UTransportIceoryx2 {`
`31`	`31`	`}`
`32`	`32`	`}`
`33`	`33`
`34`		`- fn build_publish_subscribe() -> Arc<impl UTransport> {`
	`34`	`+ fn build_publish_subscribe() -> Arc<Iceoryx2PubSub> {`
`35`	`35`	`Iceoryx2PubSub::new()`
`36`	`36`	`}`
`37`	`37`	`}`