Run cargo fmt

Author: phil-opp

binaries/cli/src/command/build/distributed.rs

@@ -32,7 +32,8 @@ pub async fn build_distributed_dataflow(
             local_working_dir,
             uv,
         },
-    )).await?;
+    ))
+    .await?;
     eprintln!("dataflow build triggered: {build_id}");
     Ok(build_id)
 }

binaries/cli/src/command/build/local.rs

@@ -17,13 +17,7 @@ pub async fn build_dataflow_locally(
     working_dir: PathBuf,
     uv: bool,
 ) -> eyre::Result<BuildInfo> {
-    build_dataflow(
-        dataflow,
-        git_sources,
-        dataflow_session,
-        working_dir,
-        uv,
-    ).await
+    build_dataflow(dataflow, git_sources, dataflow_session, working_dir, uv).await
 }
 
 async fn build_dataflow(

binaries/cli/src/command/destroy.rs

@@ -23,6 +23,7 @@ impl Executable for Destroy {
         up::destroy(
             self.config.as_deref(),
             (self.coordinator_addr, self.coordinator_port).into(),
-        ).await
+        )
+        .await
     }
 }

binaries/cli/src/command/inspect/top.rs

@@ -5,7 +5,9 @@ use std::{
 
 use clap::Args;
 use crossterm::{
-    event::{self, DisableMouseCapture, EnableMouseCapture, Event, KeyCode, KeyEvent, KeyEventKind},
+    event::{
+        self, DisableMouseCapture, EnableMouseCapture, Event, KeyCode, KeyEvent, KeyEventKind,
+    },
     execute,
     terminal::{EnterAlternateScreen, LeaveAlternateScreen, disable_raw_mode, enable_raw_mode},
 };
@@ -67,7 +69,8 @@ impl Executable for Top {
             self.coordinator_addr,
             self.coordinator_port,
             refresh_duration,
-        ).await;
+        )
+        .await;
 
         // Restore terminal
         disable_raw_mode()?;

binaries/cli/src/command/logs.rs

@@ -46,15 +46,17 @@ impl Executable for LogsArgs {
         let client = connect_to_coordinator_rpc(self.coordinator_addr, self.coordinator_port)
             .await
             .wrap_err("failed to connect to dora coordinator")?;
-        let uuid = resolve_dataflow_identifier_interactive(&client, self.dataflow.as_deref()).await?;
+        let uuid =
+            resolve_dataflow_identifier_interactive(&client, self.dataflow.as_deref()).await?;
         logs(
             &client,
             uuid,
             self.node,
             self.tail,
             self.follow,
             (self.coordinator_addr, self.coordinator_port).into(),
-        ).await
+        )
+        .await
     }
 }
 
@@ -72,7 +74,8 @@ pub async fn logs(
         None,
         node.to_string(),
         tail,
-    )).await?;
+    ))
+    .await?;
 
     std::io::stdout()
         .write_all(&logs)

binaries/cli/src/command/stop.rs

@@ -58,7 +58,9 @@ impl Executable for Stop {
             (None, Some(name)) => {
                 stop_dataflow_by_name(name, self.grace_duration, self.force, &client).await
             }
-            (None, None) => stop_dataflow_interactive(self.grace_duration, self.force, &client).await,
+            (None, None) => {
+                stop_dataflow_interactive(self.grace_duration, self.force, &client).await
+            }
         }
     }
 }
@@ -68,7 +70,9 @@ async fn stop_dataflow_interactive(
     force: bool,
     client: &CliControlClient,
 ) -> eyre::Result<()> {
-    let list = query_running_dataflows(client).await.wrap_err("failed to query running dataflows")?;
+    let list = query_running_dataflows(client)
+        .await
+        .wrap_err("failed to query running dataflows")?;
     let active = list.get_active();
     if active.is_empty() {
         eprintln!("No dataflows are running");

binaries/cli/src/command/system/status.rs

@@ -27,29 +27,30 @@ pub async fn check_environment(coordinator_addr: SocketAddr) -> eyre::Result<()>
     let mut stdout = termcolor::StandardStream::stdout(color_choice);
 
     // Coordinator status
-    let client = match connect_to_coordinator_rpc(coordinator_addr.ip(), coordinator_addr.port()).await {
-        Ok(client) => {
-            let _ = stdout.set_color(ColorSpec::new().set_fg(Some(Color::Green)));
-            write!(stdout, "✓ ")?;
-            let _ = stdout.reset();
-            writeln!(stdout, "Coordinator: Running")?;
-            writeln!(
-                stdout,
-                "  Address: {}:{}",
-                coordinator_addr.ip(),
-                coordinator_addr.port()
-            )?;
-            Some(client)
-        }
-        Err(_) => {
-            let _ = stdout.set_color(ColorSpec::new().set_fg(Some(Color::Red)));
-            write!(stdout, "✗ ")?;
-            let _ = stdout.reset();
-            writeln!(stdout, "Coordinator: Not running")?;
-            error_occurred = true;
-            None
-        }
-    };
+    let client =
+        match connect_to_coordinator_rpc(coordinator_addr.ip(), coordinator_addr.port()).await {
+            Ok(client) => {
+                let _ = stdout.set_color(ColorSpec::new().set_fg(Some(Color::Green)));
+                write!(stdout, "✓ ")?;
+                let _ = stdout.reset();
+                writeln!(stdout, "Coordinator: Running")?;
+                writeln!(
+                    stdout,
+                    "  Address: {}:{}",
+                    coordinator_addr.ip(),
+                    coordinator_addr.port()
+                )?;
+                Some(client)
+            }
+            Err(_) => {
+                let _ = stdout.set_color(ColorSpec::new().set_fg(Some(Color::Red)));
+                write!(stdout, "✗ ")?;
+                let _ = stdout.reset();
+                writeln!(stdout, "Coordinator: Not running")?;
+                error_occurred = true;
+                None
+            }
+        };
 
     // Daemon status
     let daemon_running_result = match client.as_ref() {
@@ -131,7 +132,9 @@ impl Executable for Status {
                 Descriptor::blocking_read(&dataflow)?.check(&working_dir)?;
                 check_environment((self.coordinator_addr, self.coordinator_port).into()).await?
             }
-            None => check_environment((self.coordinator_addr, self.coordinator_port).into()).await?,
+            None => {
+                check_environment((self.coordinator_addr, self.coordinator_port).into()).await?
+            }
         }
 
         Ok(())

binaries/cli/src/command/topic/info.rs

@@ -150,11 +150,7 @@ async fn info(
         .await
         .context("failed to open zenoh session")?;
 
-    let subscribe_topic = zenoh_output_publish_topic(
-        dataflow_id,
-        &topic.node_id,
-        &topic.data_id,
-    );
+    let subscribe_topic = zenoh_output_publish_topic(dataflow_id, &topic.node_id, &topic.data_id);
     let subscriber = zenoh_session
         .declare_subscriber(subscribe_topic)
         .await
@@ -168,19 +164,18 @@ async fn info(
 
     // Collect messages for the specified duration
     while Instant::now() < end_time {
-        let Ok(sample) = tokio::time::timeout_at(deadline, subscriber.recv_async()).await
-        else {
+        let Ok(sample) = tokio::time::timeout_at(deadline, subscriber.recv_async()).await else {
             break;
         };
 
         match sample {
             Ok(sample) => {
-                let event = match Timestamped::deserialize_inter_daemon_event(
-                    &sample.payload().to_bytes(),
-                ) {
-                    Ok(event) => event,
-                    Err(_) => continue,
-                };
+                let event =
+                    match Timestamped::deserialize_inter_daemon_event(&sample.payload().to_bytes())
+                    {
+                        Ok(event) => event,
+                        Err(_) => continue,
+                    };
 
                 match event.inner {
                     InterDaemonEvent::Output { metadata, data, .. } => {