Squash async worker changes

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

Author: mxgrey

examples/minimal_client_service/src/minimal_client.rs

@@ -8,29 +8,27 @@ fn main() -> Result<(), Error> {
 
     let client = node.create_client::<example_interfaces::srv::AddTwoInts>("add_two_ints")?;
 
-    let request = example_interfaces::srv::AddTwoInts_Request { a: 41, b: 1 };
-
     println!("Starting client");
 
     while !client.service_is_ready()? {
         std::thread::sleep(std::time::Duration::from_millis(10));
     }
 
-    client.async_send_request_with_callback(
-        &request,
-        move |response: example_interfaces::srv::AddTwoInts_Response| {
-            println!(
-                "Result of {} + {} is: {}",
-                request.a, request.b, response.sum
-            );
-        },
-    )?;
+    let request = example_interfaces::srv::AddTwoInts_Request { a: 41, b: 1 };
+
+    let response: Promise<example_interfaces::srv::AddTwoInts_Response> = client.call(&request).unwrap();
 
-    std::thread::sleep(std::time::Duration::from_millis(500));
+    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::default())
-        .first_error()
-        .map_err(|err| err.into())
+        .spin(SpinOptions::new().until_promise_resolved(promise))
+        .first_error()?;
+    Ok(())
 }

examples/minimal_client_service/src/minimal_client_async.rs

@@ -17,18 +17,19 @@ async fn main() -> Result<(), Error> {
 
     let request = example_interfaces::srv::AddTwoInts_Request { a: 41, b: 1 };
 
-    let future = client.call_async(&request);
+    let promise = client.call_then(
+        &request,
+        move |response: example_interfaces::srv::AddTwoInts_Response| {
+            println!(
+                "Result of {} + {} is: {}",
+                request.a, request.b, response.sum,
+            );
+        }
+    ).unwrap();
 
     println!("Waiting for response");
-
-    let rclrs_spin = tokio::task::spawn_blocking(move || executor.spin(SpinOptions::default()));
-
-    let response = future.await?;
-    println!(
-        "Result of {} + {} is: {}",
-        request.a, request.b, response.sum
-    );
-
-    rclrs_spin.await.ok();
+    executor
+        .spin(SpinOptions::new().until_promise_resolved(promise))
+        .first_error()?;
     Ok(())
 }

examples/minimal_client_service/src/minimal_service.rs

@@ -2,10 +2,15 @@ use anyhow::{Error, Result};
 use rclrs::*;
 
 fn handle_service(
-    _request_header: &rclrs::rmw_request_id_t,
     request: example_interfaces::srv::AddTwoInts_Request,
+    info: ServiceInfo,
 ) -> example_interfaces::srv::AddTwoInts_Response {
-    println!("request: {} + {}", request.a, request.b);
+    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 {
         sum: request.a + request.b,
     }

examples/minimal_pub_sub/src/minimal_subscriber.rs

@@ -7,19 +7,19 @@ fn main() -> Result<(), Error> {
 
     let node = executor.create_node("minimal_subscriber")?;
 
-    let mut num_messages: usize = 0;
-
-    let _subscription = node.create_subscription::<std_msgs::msg::String, _>(
+    let worker = node.create_worker::<usize>(0);
+    let _subscription = worker.create_subscription::<std_msgs::msg::String, _>(
         "topic",
-        move |msg: std_msgs::msg::String| {
-            num_messages += 1;
+        move |num_messages: &mut usize, msg: std_msgs::msg::String| {
+            *num_messages += 1;
             println!("I heard: '{}'", msg.data);
-            println!("(Got {} messages so far)", num_messages);
+            println!("(Got {} messages so far)", *num_messages);
         },
     )?;
 
+    println!("Waiting for messages...");
     executor
         .spin(SpinOptions::default())
-        .first_error()
-        .map_err(|err| err.into())
+        .first_error()?;
+    Ok(())
 }

examples/minimal_pub_sub/src/minimal_two_nodes.rs

@@ -1,47 +1,43 @@
 use rclrs::*;
-use std::sync::{
-    atomic::{AtomicU32, Ordering},
-    Arc, Mutex,
-};
+use std::sync::Arc;
 
 use anyhow::{Error, Result};
 
-struct MinimalSubscriber {
-    num_messages: AtomicU32,
+struct MinimalSubscriberNode {
+    #[allow(unused)]
+    subscription: WorkerSubscription<std_msgs::msg::String, SubscriptionData>,
+}
+
+struct SubscriptionData {
     node: Node,
-    subscription: Mutex<Option<Subscription<std_msgs::msg::String>>>,
+    num_messages: usize,
 }
 
-impl MinimalSubscriber {
-    pub fn new(executor: &Executor, name: &str, topic: &str) -> Result<Arc<Self>, RclrsError> {
+impl MinimalSubscriberNode {
+    pub fn new(executor: &Executor, name: &str, topic: &str) -> Result<Self, RclrsError> {
         let node = executor.create_node(name)?;
-        let minimal_subscriber = Arc::new(MinimalSubscriber {
-            num_messages: 0.into(),
-            node,
-            subscription: None.into(),
-        });
-
-        let minimal_subscriber_aux = Arc::clone(&minimal_subscriber);
-        let subscription = minimal_subscriber
-            .node
-            .create_subscription::<std_msgs::msg::String, _>(
-                topic,
-                move |msg: std_msgs::msg::String| {
-                    minimal_subscriber_aux.callback(msg);
-                },
-            )?;
-        *minimal_subscriber.subscription.lock().unwrap() = Some(subscription);
-        Ok(minimal_subscriber)
-    }
 
-    fn callback(&self, msg: std_msgs::msg::String) {
-        self.num_messages.fetch_add(1, Ordering::SeqCst);
-        println!("[{}] I heard: '{}'", self.node.name(), msg.data);
-        println!(
-            "[{}] (Got {} messages so far)",
-            self.node.name(),
-            self.num_messages.load(Ordering::SeqCst)
+        let worker = node.create_worker::<SubscriptionData>(
+            SubscriptionData {
+                node: Arc::clone(&node),
+                num_messages: 0,
+            }
         );
+
+        let subscription = worker.create_subscription(
+            topic,
+            |data: &mut SubscriptionData, msg: std_msgs::msg::String| {
+                data.num_messages += 1;
+                println!("[{}] I heard: '{}'", data.node.name(), msg.data);
+                println!(
+                    "[{}] (Got {} messages so far)",
+                    data.node.name(),
+                    data.num_messages,
+                );
+            }
+        )?;
+
+        Ok(MinimalSubscriberNode { subscription })
     }
 }
 
@@ -50,13 +46,15 @@ fn main() -> Result<(), Error> {
     let publisher_node = executor.create_node("minimal_publisher")?;
 
     let _subscriber_node_one =
-        MinimalSubscriber::new(&executor, "minimal_subscriber_one", "topic")?;
+        MinimalSubscriberNode::new(&executor, "minimal_subscriber_one", "topic")?;
     let _subscriber_node_two =
-        MinimalSubscriber::new(&executor, "minimal_subscriber_two", "topic")?;
+        MinimalSubscriberNode::new(&executor, "minimal_subscriber_two", "topic")?;
 
     let publisher = publisher_node.create_publisher::<std_msgs::msg::String>("topic")?;
 
-    std::thread::spawn(move || -> Result<(), RclrsError> {
+    // TODO(@mxgrey): Replace this with a timer once we have the Timer feature
+    // merged in.
+    std::thread::spawn(move || -> Result<(), rclrs::RclrsError> {
         let mut message = std_msgs::msg::String::default();
         let mut publish_count: u32 = 1;
         loop {
@@ -69,7 +67,8 @@ fn main() -> Result<(), Error> {
     });
 
     executor
-        .spin(SpinOptions::default())
-        .first_error()
-        .map_err(|err| err.into())
+        .spin(rclrs::SpinOptions::default())
+        .first_error()?;
+    Ok(())
 }
+

examples/minimal_pub_sub/src/zero_copy_subscriber.rs

@@ -6,19 +6,18 @@ fn main() -> Result<(), Error> {
 
     let node = executor.create_node("minimal_subscriber")?;
 
-    let mut num_messages: usize = 0;
-
-    let _subscription = node.create_subscription::<std_msgs::msg::UInt32, _>(
+    let worker = node.create_worker::<usize>(0);
+    let _subscription = worker.create_subscription::<std_msgs::msg::UInt32, _>(
         "topic",
-        move |msg: rclrs::ReadOnlyLoanedMessage<'_, std_msgs::msg::UInt32>| {
-            num_messages += 1;
+        move |num_messages: &mut usize, msg: ReadOnlyLoanedMessage<std_msgs::msg::UInt32>| {
+            *num_messages += 1;
             println!("I heard: '{}'", msg.data);
-            println!("(Got {} messages so far)", num_messages);
+            println!("(Got {} messages so far)", *num_messages);
         },
     )?;
 
     executor
         .spin(SpinOptions::default())
-        .first_error()
-        .map_err(|err| err.into())
+        .first_error()?;
+    Ok(())
 }

examples/rust_pubsub/src/simple_publisher.rs

@@ -2,14 +2,16 @@ use rclrs::*;
 use std::{thread, time::Duration};
 use std_msgs::msg::String as StringMsg;
 
-struct SimplePublisher {
+struct SimplePublisherNode {
     publisher: Publisher<StringMsg>,
 }
 
-impl SimplePublisher {
+impl SimplePublisherNode {
     fn new(executor: &Executor) -> Result<Self, RclrsError> {
         let node = executor.create_node("simple_publisher").unwrap();
-        let publisher = node.create_publisher("publish_hello").unwrap();
+        let publisher = node
+            .create_publisher("publish_hello")
+            .unwrap();
         Ok(Self { publisher })
     }
 
@@ -24,11 +26,14 @@ impl SimplePublisher {
 
 fn main() -> Result<(), RclrsError> {
     let mut executor = Context::default_from_env().unwrap().create_basic_executor();
-    let publisher = SimplePublisher::new(&executor).unwrap();
+    let node = SimplePublisherNode::new(&executor).unwrap();
     let mut count: i32 = 0;
+
+    // TODO(@mxgrey): Replace this with a timer once the Timer feature
+    // is merged.
     thread::spawn(move || loop {
         thread::sleep(Duration::from_millis(1000));
-        count = publisher.publish_data(count).unwrap();
+        count = node.publish_data(count).unwrap();
     });
     executor.spin(SpinOptions::default()).first_error()
 }

examples/rust_pubsub/src/simple_subscriber.rs

@@ -7,7 +7,8 @@ use std::{
 use std_msgs::msg::String as StringMsg;
 
 pub struct SimpleSubscriptionNode {
-    _subscriber: Subscription<StringMsg>,
+    #[allow(unused)]
+    subscriber: Subscription<StringMsg>,
     data: Arc<Mutex<Option<StringMsg>>>,
 }
 
@@ -16,12 +17,15 @@ impl SimpleSubscriptionNode {
         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
-            .create_subscription::<StringMsg, _>("publish_hello", move |msg: StringMsg| {
-                *data_mut.lock().unwrap() = Some(msg);
-            })
+        let subscriber = node
+            .create_subscription::<StringMsg, _>(
+                "publish_hello",
+                move |msg: StringMsg| {
+                    *data_mut.lock().unwrap() = Some(msg);
+                },
+            )
             .unwrap();
-        Ok(Self { _subscriber, data })
+        Ok(Self { subscriber, data })
     }
     fn data_callback(&self) -> Result<(), RclrsError> {
         if let Some(data) = self.data.lock().unwrap().as_ref() {
@@ -34,10 +38,13 @@ impl SimpleSubscriptionNode {
 }
 fn main() -> Result<(), RclrsError> {
     let mut executor = Context::default_from_env().unwrap().create_basic_executor();
-    let subscription = SimpleSubscriptionNode::new(&executor).unwrap();
+    let node = SimpleSubscriptionNode::new(&executor).unwrap();
+
+    // TODO(@mxgrey): Replace this thread with a timer when the Timer feature
+    // gets merged.
     thread::spawn(move || loop {
         thread::sleep(Duration::from_millis(1000));
-        subscription.data_callback().unwrap()
+        node.data_callback().unwrap()
     });
     executor.spin(SpinOptions::default()).first_error()
 }

examples/worker_demo/Cargo.toml

@@ -0,0 +1,8 @@
+[package]
+name = "worker_demo"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+rclrs = "0.4"
+std_msgs = "*"