Clean up examples using new API

Signed-off-by: Michael X. Grey <[email protected]>

Author: mxgrey

examples/minimal_client_service/src/minimal_client.rs

@@ -1,32 +1,30 @@
 use anyhow::{Error, Result};
 use rclrs::*;
+use example_interfaces::srv::*;
 
 fn main() -> Result<(), Error> {
     let mut executor = Context::default_from_env()?.create_basic_executor();
 
     let node = executor.create_node("minimal_client")?;
 
-    let client = node.create_client::<example_interfaces::srv::AddTwoInts>("add_two_ints")?;
+    let client = node.create_client::<AddTwoInts>("add_two_ints")?;
 
-    println!("Starting client");
-
-    while !client.service_is_ready()? {
-        std::thread::sleep(std::time::Duration::from_millis(10));
-    }
+    let promise = executor.commands().run(async move {
+        println!("Waiting for service...");
+        client.notify_on_service_ready().await.unwrap();
 
-    let request = example_interfaces::srv::AddTwoInts_Request { a: 41, b: 1 };
+        let request = AddTwoInts_Request { a: 41, b: 1 };
 
-    let response: Promise<example_interfaces::srv::AddTwoInts_Response> = client.call(&request).unwrap();
+        println!("Waiting for response");
+        let response: AddTwoInts_Response =
+            client.call(&request).unwrap().await.unwrap();
 
-    let promise = executor.commands().run(async move {
-        let response = response.await.unwrap();
         println!(
             "Result of {} + {} is: {}",
             request.a, request.b, response.sum,
         );
     });
 
-    println!("Waiting for response");
     executor
         .spin(SpinOptions::new().until_promise_resolved(promise))
         .first_error()?;

examples/minimal_client_service/src/minimal_client_async.rs

@@ -1,25 +1,26 @@
 use anyhow::{Error, Result};
 use rclrs::*;
+use example_interfaces::srv::*;
 
 #[tokio::main]
 async fn main() -> Result<(), Error> {
     let mut executor = Context::default_from_env()?.create_basic_executor();
 
     let node = executor.create_node("minimal_client")?;
 
-    let client = node.create_client::<example_interfaces::srv::AddTwoInts>("add_two_ints")?;
+    let client = node.create_client::<AddTwoInts>("add_two_ints")?;
 
     println!("Starting client");
 
     while !client.service_is_ready()? {
         std::thread::sleep(std::time::Duration::from_millis(10));
     }
 
-    let request = example_interfaces::srv::AddTwoInts_Request { a: 41, b: 1 };
+    let request = AddTwoInts_Request { a: 41, b: 1 };
 
     let promise = client.call_then(
         &request,
-        move |response: example_interfaces::srv::AddTwoInts_Response| {
+        move |response: AddTwoInts_Response| {
             println!(
                 "Result of {} + {} is: {}",
                 request.a, request.b, response.sum,

examples/minimal_client_service/src/minimal_service.rs

@@ -1,17 +1,18 @@
 use anyhow::{Error, Result};
 use rclrs::*;
+use example_interfaces::srv::*;
 
 fn handle_service(
-    request: example_interfaces::srv::AddTwoInts_Request,
+    request: AddTwoInts_Request,
     info: ServiceInfo,
-) -> example_interfaces::srv::AddTwoInts_Response {
+) -> AddTwoInts_Response {
     let timestamp = info
         .received_timestamp
         .map(|t| format!(" at [{t:?}]"))
         .unwrap_or(String::new());
 
     println!("request{timestamp}: {} + {}", request.a, request.b);
-    example_interfaces::srv::AddTwoInts_Response {
+    AddTwoInts_Response {
         sum: request.a + request.b,
     }
 }
@@ -22,11 +23,11 @@ fn main() -> Result<(), Error> {
     let node = executor.create_node("minimal_service")?;
 
     let _server = node
-        .create_service::<example_interfaces::srv::AddTwoInts, _>("add_two_ints", handle_service)?;
+        .create_service::<AddTwoInts, _>("add_two_ints", handle_service)?;
 
     println!("Starting server");
     executor
         .spin(SpinOptions::default())
-        .first_error()
-        .map_err(|err| err.into())
+        .first_error()?;
+    Ok(())
 }

examples/rust_pubsub/src/simple_subscriber.rs

@@ -1,39 +1,31 @@
 use rclrs::*;
 use std::{
-    sync::{Arc, Mutex},
     thread,
     time::Duration,
 };
 use std_msgs::msg::String as StringMsg;
 
 pub struct SimpleSubscriptionNode {
     #[allow(unused)]
-    subscriber: Subscription<StringMsg>,
-    data: Arc<Mutex<Option<StringMsg>>>,
+    subscriber: WorkerSubscription<StringMsg, Option<String>>,
+    worker: Worker<Option<String>>,
 }
 
 impl SimpleSubscriptionNode {
     fn new(executor: &Executor) -> Result<Self, RclrsError> {
         let node = executor.create_node("simple_subscription").unwrap();
-        let data: Arc<Mutex<Option<StringMsg>>> = Arc::new(Mutex::new(None));
-        let data_mut: Arc<Mutex<Option<StringMsg>>> = Arc::clone(&data);
-        let subscriber = node
+        let worker = node.create_worker(None);
+
+        let subscriber = worker
             .create_subscription::<StringMsg, _>(
                 "publish_hello",
-                move |msg: StringMsg| {
-                    *data_mut.lock().unwrap() = Some(msg);
+                move |data: &mut Option<String>, msg: StringMsg| {
+                    *data = Some(msg.data);
                 },
             )
             .unwrap();
-        Ok(Self { subscriber, data })
-    }
-    fn data_callback(&self) -> Result<(), RclrsError> {
-        if let Some(data) = self.data.lock().unwrap().as_ref() {
-            println!("{}", data.data);
-        } else {
-            println!("No message available yet.");
-        }
-        Ok(())
+
+        Ok(Self { subscriber, worker })
     }
 }
 fn main() -> Result<(), RclrsError> {
@@ -43,8 +35,15 @@ fn main() -> Result<(), RclrsError> {
     // TODO(@mxgrey): Replace this thread with a timer when the Timer feature
     // gets merged.
     thread::spawn(move || loop {
-        thread::sleep(Duration::from_millis(1000));
-        node.data_callback().unwrap()
+        thread::sleep(Duration::from_secs(1));
+        let _ = node.worker.run(|data: &mut Option<String>| {
+            if let Some(data) = data {
+                println!("{data}");
+            } else {
+                println!("No message available yet.");
+            }
+        });
     });
+
     executor.spin(SpinOptions::default()).first_error()
 }