Implement PR feedback

Author: marc-sift

rust/crates/sift_cli/src/cmd/test_server/metrics_streaming_client.rs

@@ -2,9 +2,9 @@ use super::Context;
 use anyhow::{Ok, anyhow};
 use crossterm::style::Stylize;
 use sift_stream::{
-    ChannelConfig, ChannelDataType, ChannelValue, Credentials, Flow, FlowBuilder, FlowConfig,
-    IngestionConfigEncoder, IngestionConfigForm, RecoveryStrategy, RetryPolicy, RunForm,
-    SiftStream, SiftStreamBuilder, TimeValue,
+    ChannelConfig, ChannelDataType, ChannelValue, Credentials, Flow, FlowConfig,
+    IngestionConfigEncoder, IngestionConfigForm, RecoveryStrategy, RetryPolicy, SiftStream,
+    SiftStreamBuilder, TimeValue,
 };
 
 /// Streams metrics to Sift.
@@ -32,7 +32,7 @@ impl MetricsStreamingClient {
         };
 
         Ok(Some(MetricsStreamingClient {
-            ctx: ctx,
+            ctx,
             asset_name: asset_name.clone(),
             sift_stream: None,
         }))

rust/crates/sift_cli/src/cmd/test_server/mod.rs

@@ -1,7 +1,7 @@
 use super::Context;
 use crate::cmd::test_server::metrics_streaming_client::MetricsStreamingClient;
 use crate::{cli::TestServerArgs, util::tty::Output};
-use anyhow::Result;
+use anyhow::{Context as AnyhowContext, Result};
 use server::TestServer;
 use sift_rs::assets::v1::asset_service_server::AssetServiceServer;
 use sift_rs::ingest::v1::ingest_service_server::IngestServiceServer;
@@ -20,14 +20,23 @@ pub mod server;
 use crate::cmd::test_server::metrics_streaming_client::Metrics;
 
 pub async fn run(ctx: Context, args: TestServerArgs) -> Result<ExitCode> {
-    let local_address = args.local_address.unwrap_or("127.0.0.1:50051".into());
-    let addr = local_address.parse()?;
+    let local_address = args
+        .local_address
+        .unwrap_or_else(|| "0.0.0.0:50051".to_string());
+    let addr = local_address
+        .parse()
+        .context(format!("failed to parse local_address: {}", local_address))?;
 
     // Initialize streaming client.
     let mut streaming_client =
-        MetricsStreamingClient::build(ctx, &args.stream_metrics, &args.metrics_asset_name)?;
-    if streaming_client.is_some() {
-        streaming_client.as_mut().unwrap().initialize().await?;
+        MetricsStreamingClient::build(ctx, &args.stream_metrics, &args.metrics_asset_name)
+            .context("failed to create metrics streaming client")?;
+
+    if let Some(client) = streaming_client.as_mut() {
+        client
+            .initialize()
+            .await
+            .context("failed to initialize streaming client")?;
     }
 
     // Channel to signal program exit.
@@ -54,27 +63,25 @@ pub async fn run(ctx: Context, args: TestServerArgs) -> Result<ExitCode> {
 
     // Start task to ingest metrics to Sift.
     let ingest_metrics_task = tokio::spawn(async move {
-        if streaming_client.is_none() {
-            return;
-        }
-
-        let mut client = streaming_client.unwrap();
-        loop {
-            tokio::select! {
-                _ = shutdown_rx2.changed() => {
-                    Output::new().line("Ingest task shutting down").print();
-                    break;
-                }
-                Some(metrics) = metrics_rx.recv() => {
-                    client.ingest(metrics).await;
-                }
-            };
+        if let Some(client) = streaming_client.as_mut() {
+            loop {
+                tokio::select! {
+                    _ = shutdown_rx2.changed() => {
+                        Output::new().line("Ingest task shutting down").print();
+                        break;
+                    }
+                    Some(metrics) = metrics_rx.recv() => {
+                        client.ingest(metrics).await;
+                    }
+                };
+            }
         }
     });
 
     let reflection_service = Builder::configure()
         .register_encoded_file_descriptor_set(FILE_DESCRIPTOR_SET)
-        .build_v1()?;
+        .build_v1()
+        .context("failed to create gRPC reflection service")?;
 
     Output::new()
         .line(format!("Server listening on {addr}"))
@@ -92,8 +99,12 @@ pub async fn run(ctx: Context, args: TestServerArgs) -> Result<ExitCode> {
         })
         .await?;
 
-    calc_stats_task.await?;
-    ingest_metrics_task.await?;
+    calc_stats_task
+        .await
+        .context("failed to await calculation task")?;
+    ingest_metrics_task
+        .await
+        .context("failed to await ingestion task")?;
 
     Output::new().line("Exiting.").print();
 

rust/crates/sift_cli/src/cmd/test_server/server.rs

@@ -58,6 +58,8 @@ impl TestServer {
         metrics_tx: Sender<Metrics>,
         streaming_enabled: bool,
     ) {
+        let mut stdout = stdout();
+
         let mut last_total_num_bytes_read: u64 = 0;
         let mut last_total_num_messages: u64 = 0;
 
@@ -79,12 +81,12 @@ impl TestServer {
                     last_total_num_messages = current_total_num_messages;
 
                     // Clear terminal and print metrics.
-                    stdout()
+                    stdout
                         .execute(terminal::Clear(terminal::ClearType::All))
                         .expect("");
-                    stdout().execute(cursor::MoveTo(0, 0)).expect("msg");
-                    stdout().execute(cursor::MoveUp(5)).expect("terminal error");
-                    stdout().execute(terminal::Clear(terminal::ClearType::FromCursorDown)).expect("msg");
+                    stdout.execute(cursor::MoveTo(0, 0)).expect("msg");
+                    stdout.execute(cursor::MoveUp(5)).expect("terminal error");
+                    stdout.execute(terminal::Clear(terminal::ClearType::FromCursorDown)).expect("msg");
 
                     Output::new().line(format!("Total num streams:  {current_total_num_streams}")).print();
                     Output::new().line(format!("Total num bytes:    {current_total_num_bytes_read}")).print();
@@ -115,11 +117,7 @@ impl TestServer {
 #[tonic::async_trait]
 impl PingService for TestServer {
     async fn ping(&self, _request: Request<PingRequest>) -> Result<Response<PingResponse>, Status> {
-        let resp = PingResponse {
-            response: "".into(),
-        };
-
-        Ok(Response::new(resp))
+        Ok(Response::new(PingResponse::default()))
     }
 }