Fix formatting

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

Author: mxgrey

rclrs/src/client.rs

@@ -8,9 +8,9 @@ use std::{
 use rosidl_runtime_rs::Message;
 
 use crate::{
-    error::ToResult, rcl_bindings::*, IntoPrimitiveOptions, MessageCow, Node, Promise, QoSProfile,
-    RclPrimitive, RclPrimitiveHandle, RclPrimitiveKind, RclReturnCode, RclrsError, ServiceInfo,
-    Waitable, WaitableLifecycle, ENTITY_LIFECYCLE_MUTEX, log_fatal,
+    error::ToResult, log_fatal, rcl_bindings::*, IntoPrimitiveOptions, MessageCow, Node, Promise,
+    QoSProfile, RclPrimitive, RclPrimitiveHandle, RclPrimitiveKind, RclReturnCode, RclrsError,
+    ServiceInfo, Waitable, WaitableLifecycle, ENTITY_LIFECYCLE_MUTEX,
 };
 
 mod client_async_callback;
@@ -284,9 +284,9 @@ where
     /// `until_promise_resolved` in [`SpinOptions`][crate::SpinOptions].
     pub fn notify_on_service_ready(self: &Arc<Self>) -> Promise<()> {
         let client = Arc::clone(self);
-        self.handle.node.notify_on_graph_change(
-            move || client.service_is_ready().is_ok_and(|r| r),
-        )
+        self.handle
+            .node
+            .notify_on_graph_change(move || client.service_is_ready().is_ok_and(|r| r))
     }
 
     /// Get the name of the service that this client intends to call.
@@ -598,4 +598,3 @@ mod tests {
         Ok(())
     }
 }
-

rclrs/src/context.rs

@@ -6,10 +6,7 @@ use std::{
     vec::Vec,
 };
 
-use crate::{
-    rcl_bindings::*, Executor, ExecutorRuntime, LoggingLifecycle, RclrsError,
-    ToResult,
-};
+use crate::{rcl_bindings::*, Executor, ExecutorRuntime, LoggingLifecycle, RclrsError, ToResult};
 
 /// This is locked whenever initializing or dropping any middleware entity
 /// because we have found issues in RCL and some RMW implementations that

rclrs/src/error.rs

@@ -113,16 +113,10 @@ impl Display for RclrsError {
                 )
             }
             RclrsError::ParameterDeclarationError(err) => {
-                write!(
-                    f,
-                    "An error occurred while declaring a parameter: {err}",
-                )
+                write!(f, "An error occurred while declaring a parameter: {err}",)
             }
             RclrsError::PoisonedMutex => {
-                write!(
-                    f,
-                    "A mutex used internally has been poisoned"
-                )
+                write!(f, "A mutex used internally has been poisoned")
             }
         }
     }

rclrs/src/executor.rs

@@ -1,7 +1,10 @@
 mod basic_executor;
 pub use self::basic_executor::*;
 
-use crate::{WeakActivityListener, Context, ContextHandle, GuardCondition, IntoNodeOptions, Node, RclrsError, Waitable};
+use crate::{
+    Context, ContextHandle, GuardCondition, IntoNodeOptions, Node, RclrsError, Waitable,
+    WeakActivityListener,
+};
 pub use futures::channel::oneshot::Receiver as Promise;
 use futures::{
     channel::oneshot,
@@ -92,7 +95,9 @@ impl Executor {
     {
         let executor_channel = runtime.channel();
         let async_worker_commands = ExecutorCommands::impl_create_worker_commands(
-            &Context { handle: Arc::clone(&context) },
+            &Context {
+                handle: Arc::clone(&context),
+            },
             &*executor_channel,
             Box::new(()),
         );
@@ -168,16 +173,18 @@ impl ExecutorCommands {
         F::Output: Send,
     {
         let (mut sender, receiver) = oneshot::channel();
-        self.async_worker_commands.channel.add_async_task(Box::pin(async move {
-            let cancellation = sender.cancellation();
-            let output = match select(cancellation, std::pin::pin!(f)).await {
-                // The task was cancelled
-                Either::Left(_) => return,
-                // The task completed
-                Either::Right((output, _)) => output,
-            };
-            sender.send(output).ok();
-        }));
+        self.async_worker_commands
+            .channel
+            .add_async_task(Box::pin(async move {
+                let cancellation = sender.cancellation();
+                let output = match select(cancellation, std::pin::pin!(f)).await {
+                    // The task was cancelled
+                    Either::Left(_) => return,
+                    // The task completed
+                    Either::Right((output, _)) => output,
+                };
+                sender.send(output).ok();
+            }));
 
         receiver
     }
@@ -203,9 +210,11 @@ impl ExecutorCommands {
         F::Output: Send,
     {
         let (sender, receiver) = oneshot::channel();
-        self.async_worker_commands.channel.add_async_task(Box::pin(async move {
-            sender.send(f.await).ok();
-        }));
+        self.async_worker_commands
+            .channel
+            .add_async_task(Box::pin(async move {
+                sender.send(f.await).ok();
+            }));
         receiver
     }
 
@@ -244,7 +253,10 @@ impl ExecutorCommands {
         &self.async_worker_commands
     }
 
-    pub(crate) fn create_worker_commands(&self, payload: Box<dyn Any + Send>) -> Arc<WorkerCommands> {
+    pub(crate) fn create_worker_commands(
+        &self,
+        payload: Box<dyn Any + Send>,
+    ) -> Arc<WorkerCommands> {
         Self::impl_create_worker_commands(&self.context, &*self.executor_channel, payload)
     }
 
@@ -344,10 +356,7 @@ pub struct ExecutorWorkerOptions {
 pub trait ExecutorChannel: Send + Sync {
     /// Create a new channel specific to a worker whose payload must be
     /// initialized with the given function.
-    fn create_worker(
-        &self,
-        options: ExecutorWorkerOptions,
-    ) -> Arc<dyn WorkerChannel>;
+    fn create_worker(&self, options: ExecutorWorkerOptions) -> Arc<dyn WorkerChannel>;
 
     /// Wake all the wait sets that are being managed by this executor. This is
     /// used to make sure they respond to [`ExecutorCommands::halt_spinning`].
@@ -477,4 +486,3 @@ impl CreateBasicExecutor for Context {
         self.create_executor(runtime)
     }
 }
-

rclrs/src/executor/basic_executor.rs

@@ -1,8 +1,11 @@
 use futures::{
-    channel::{mpsc::{UnboundedSender, UnboundedReceiver, unbounded}, oneshot},
-    future::{BoxFuture, select, select_all, Either},
-    task::{waker_ref, ArcWake},
+    channel::{
+        mpsc::{unbounded, UnboundedReceiver, UnboundedSender},
+        oneshot,
+    },
+    future::{select, select_all, BoxFuture, Either},
     stream::StreamFuture,
+    task::{waker_ref, ArcWake},
     StreamExt,
 };
 use std::{
@@ -16,9 +19,9 @@ use std::{
 };
 
 use crate::{
-    WeakActivityListener, ExecutorChannel, ExecutorRuntime, SpinConditions, WorkerChannel,
-    RclrsError, WaitSetRunner, WaitSetRunConditions, Waitable, log_warn, log_debug, log_fatal,
-    GuardCondition, ExecutorWorkerOptions, PayloadTask,
+    log_debug, log_fatal, log_warn, ExecutorChannel, ExecutorRuntime, ExecutorWorkerOptions,
+    GuardCondition, PayloadTask, RclrsError, SpinConditions, WaitSetRunConditions, WaitSetRunner,
+    Waitable, WeakActivityListener, WorkerChannel,
 };
 
 static FAILED_TO_SEND_WORKER: &'static str =
@@ -73,7 +76,11 @@ impl AllGuardConditions {
 
     fn push(&self, guard_condition: Weak<GuardCondition>) {
         let mut inner = self.inner.lock().unwrap();
-        if inner.iter().find(|other| guard_condition.ptr_eq(other)).is_some() {
+        if inner
+            .iter()
+            .find(|other| guard_condition.ptr_eq(other))
+            .is_some()
+        {
             // This guard condition is already known
             return;
         }
@@ -115,11 +122,7 @@ impl ExecutorRuntime for BasicExecutorRuntime {
         // Use this to terminate the spinning once the wait set is finished.
         let workers_finished_clone = Arc::clone(&workers_finished);
         self.task_sender.add_async_task(Box::pin(async move {
-            let workers = manage_workers(
-                new_workers,
-                all_guard_conditions,
-                conditions,
-            ).await;
+            let workers = manage_workers(new_workers, all_guard_conditions, conditions).await;
 
             if let Err(err) = worker_result_sender.send(workers) {
                 log_fatal!(
@@ -271,10 +274,7 @@ struct BasicExecutorChannel {
 }
 
 impl ExecutorChannel for BasicExecutorChannel {
-    fn create_worker(
-        &self,
-        options: ExecutorWorkerOptions,
-    ) -> Arc<dyn WorkerChannel> {
+    fn create_worker(&self, options: ExecutorWorkerOptions) -> Arc<dyn WorkerChannel> {
         let runner = WaitSetRunner::new(options);
         let waitable_sender = runner.waitable_sender();
         let payload_task_sender = runner.payload_task_sender();
@@ -387,16 +387,19 @@ async fn manage_workers(
     mut new_workers: StreamFuture<UnboundedReceiver<WaitSetRunner>>,
     all_guard_conditions: AllGuardConditions,
     conditions: WaitSetRunConditions,
-) -> (Vec<WaitSetRunner>, StreamFuture<UnboundedReceiver<WaitSetRunner>>, Vec<RclrsError>) {
-    let mut active_runners: Vec<oneshot::Receiver<(WaitSetRunner, Result<(), RclrsError>)>> = Vec::new();
+) -> (
+    Vec<WaitSetRunner>,
+    StreamFuture<UnboundedReceiver<WaitSetRunner>>,
+    Vec<RclrsError>,
+) {
+    let mut active_runners: Vec<oneshot::Receiver<(WaitSetRunner, Result<(), RclrsError>)>> =
+        Vec::new();
     let mut finished_runners: Vec<WaitSetRunner> = Vec::new();
     let mut errors: Vec<RclrsError> = Vec::new();
 
-    let add_runner = |
-        new_runner: Option<WaitSetRunner>,
-        active_runners: &mut Vec<_>,
-        finished_runners: &mut Vec<_>,
-    | {
+    let add_runner = |new_runner: Option<WaitSetRunner>,
+                      active_runners: &mut Vec<_>,
+                      finished_runners: &mut Vec<_>| {
         if let Some(runner) = new_runner {
             all_guard_conditions.push(Arc::downgrade(runner.guard_condition()));
             if conditions.halt_spinning.load(Ordering::Acquire) {
@@ -413,16 +416,10 @@ async fn manage_workers(
     add_runner(initial_worker, &mut active_runners, &mut finished_runners);
 
     while !active_runners.is_empty() {
-        let next_event = select(
-            select_all(active_runners),
-            new_workers,
-        );
+        let next_event = select(select_all(active_runners), new_workers);
 
         match next_event.await {
-            Either::Left((
-                (finished_worker, _, remaining_workers),
-                new_worker_stream,
-            )) => {
+            Either::Left(((finished_worker, _, remaining_workers), new_worker_stream)) => {
                 match finished_worker {
                     Ok((runner, result)) => {
                         finished_runners.push(runner);
@@ -443,16 +440,13 @@ async fn manage_workers(
                 active_runners = remaining_workers;
                 new_workers = new_worker_stream;
             }
-            Either::Right((
-                (new_worker, new_worker_receiver),
-                remaining_workers,
-            )) => {
+            Either::Right(((new_worker, new_worker_receiver), remaining_workers)) => {
                 active_runners = remaining_workers.into_inner();
                 add_runner(new_worker, &mut active_runners, &mut finished_runners);
                 new_workers = new_worker_receiver.into_future();
             }
         }
-    };
+    }
 
     (finished_runners, new_workers, errors)
 }

rclrs/src/node.rs

@@ -30,11 +30,11 @@ use rosidl_runtime_rs::Message;
 
 use crate::{
     rcl_bindings::*, Client, ClientOptions, ClientState, Clock, ContextHandle, ExecutorCommands,
-    LogParams, Logger, ParameterBuilder, ParameterInterface, ParameterVariant, Parameters, Promise,
-    Publisher, PublisherOptions, PublisherState, RclrsError, Service, IntoAsyncServiceCallback,
-    IntoNodeServiceCallback, ServiceOptions, ServiceState, Subscription, IntoAsyncSubscriptionCallback,
-    IntoNodeSubscriptionCallback, SubscriptionOptions, SubscriptionState, TimeSource, ToLogParams,
-    ENTITY_LIFECYCLE_MUTEX, Worker, WorkerOptions, WorkerState,
+    IntoAsyncServiceCallback, IntoAsyncSubscriptionCallback, IntoNodeServiceCallback,
+    IntoNodeSubscriptionCallback, LogParams, Logger, ParameterBuilder, ParameterInterface,
+    ParameterVariant, Parameters, Promise, Publisher, PublisherOptions, PublisherState, RclrsError,
+    Service, ServiceOptions, ServiceState, Subscription, SubscriptionOptions, SubscriptionState,
+    TimeSource, ToLogParams, Worker, WorkerOptions, WorkerState, ENTITY_LIFECYCLE_MUTEX,
 };
 
 /// A processing unit that can communicate with other nodes. See the API of
@@ -299,7 +299,9 @@ impl NodeState {
         Payload: 'static + Send + Sync,
     {
         let options = options.into();
-        let commands = self.commands.create_worker_commands(Box::new(options.payload));
+        let commands = self
+            .commands
+            .create_worker_commands(Box::new(options.payload));
         WorkerState::create(Arc::clone(self), commands)
     }
 

rclrs/src/node/graph.rs

@@ -591,4 +591,3 @@ mod tests {
         )));
     }
 }
-

rclrs/src/node/node_options.rs

@@ -476,4 +476,3 @@ impl Drop for rcl_node_options_t {
         }
     }
 }
-

rclrs/src/service.rs

@@ -1,16 +1,16 @@
 use std::{
+    any::Any,
     boxed::Box,
     ffi::{CStr, CString},
     sync::{Arc, Mutex, MutexGuard},
-    any::Any,
 };
 
 use rosidl_runtime_rs::{Message, Service as IdlService};
 
 use crate::{
-    error::ToResult, rcl_bindings::*, WorkerCommands, IntoPrimitiveOptions, NodeHandle,
+    error::ToResult, rcl_bindings::*, IntoPrimitiveOptions, MessageCow, Node, NodeHandle,
     QoSProfile, RclPrimitive, RclPrimitiveHandle, RclPrimitiveKind, RclrsError, Waitable,
-    WaitableLifecycle, ENTITY_LIFECYCLE_MUTEX, MessageCow, Node, Worker, WorkScope,
+    WaitableLifecycle, WorkScope, Worker, WorkerCommands, ENTITY_LIFECYCLE_MUTEX,
 };
 
 mod any_service_callback;

rclrs/src/service/any_service_callback.rs

@@ -1,6 +1,8 @@
 use rosidl_runtime_rs::Service;
 
-use crate::{WorkerCommands, RclrsError, ServiceHandle, NodeServiceCallback, WorkerServiceCallback};
+use crate::{
+    NodeServiceCallback, RclrsError, ServiceHandle, WorkerCommands, WorkerServiceCallback,
+};
 
 use std::{any::Any, sync::Arc};
 
@@ -40,7 +42,9 @@ impl<T: Service> From<NodeServiceCallback<T>> for AnyServiceCallback<T, ()> {
     }
 }
 
-impl<T: Service, Payload: 'static + Send> From<WorkerServiceCallback<T, Payload>> for AnyServiceCallback<T, Payload> {
+impl<T: Service, Payload: 'static + Send> From<WorkerServiceCallback<T, Payload>>
+    for AnyServiceCallback<T, Payload>
+{
     fn from(value: WorkerServiceCallback<T, Payload>) -> Self {
         AnyServiceCallback::Worker(value)
     }