feat(enterprise): improve stats streaming UX with Ctrl+C handling (#457)

Enhances the stats streaming feature to match the UX of log streaming:

- Add Ctrl+C (SIGINT) handling for graceful shutdown
- Display informative 'Streaming... (Ctrl+C to stop)' message on start
- Show 'Stopping stats stream...' message on exit
- Consistent tokio::select! pattern across all streaming functions

Applies to:
- cluster stats streaming
- database stats streaming
- node stats streaming

Addresses part of #405 - stats streaming was already implemented,
this improves the user experience.

Author: joshrotenberg

crates/redisctl/src/commands/enterprise/stats.rs

@@ -8,6 +8,7 @@ use anyhow::Context;
 use futures::StreamExt;
 use redis_enterprise::stats::StatsHandler;
 use std::time::Duration;
+use tokio::signal;
 
 use super::utils::*;
 
@@ -135,19 +136,38 @@ async fn handle_database_stats_stream(
     let stats_handler = StatsHandler::new(client);
     let mut stream = stats_handler.stream_database(database_id, Duration::from_secs(poll_interval));
 
-    while let Some(result) = stream.next().await {
-        match result {
-            Ok(stats) => {
-                let stats_json =
-                    serde_json::to_value(stats).context("Failed to serialize stats")?;
-                let data = handle_output(stats_json, output_format, query)?;
-                print_formatted_output(data, output_format)?;
-                println!(); // Separator between polls
-            }
-            Err(e) => {
-                eprintln!("Error fetching stats: {}", e);
+    println!(
+        "Streaming database {} stats (Ctrl+C to stop)...\n",
+        database_id
+    );
+
+    loop {
+        tokio::select! {
+            // Handle Ctrl+C
+            _ = signal::ctrl_c() => {
+                println!("\nStopping stats stream...");
                 break;
             }
+            // Handle next stats entry
+            result = stream.next() => {
+                match result {
+                    Some(Ok(stats)) => {
+                        let stats_json =
+                            serde_json::to_value(stats).context("Failed to serialize stats")?;
+                        let data = handle_output(stats_json, output_format, query)?;
+                        print_formatted_output(data, output_format)?;
+                        println!(); // Separator between polls
+                    }
+                    Some(Err(e)) => {
+                        eprintln!("Error fetching stats: {}", e);
+                        break;
+                    }
+                    None => {
+                        // Stream ended
+                        break;
+                    }
+                }
+            }
         }
     }
     Ok(())
@@ -226,19 +246,32 @@ async fn handle_node_stats_stream(
     let stats_handler = StatsHandler::new(client);
     let mut stream = stats_handler.stream_node(node_id, Duration::from_secs(poll_interval));
 
-    while let Some(result) = stream.next().await {
-        match result {
-            Ok(stats) => {
-                let stats_json =
-                    serde_json::to_value(stats).context("Failed to serialize stats")?;
-                let data = handle_output(stats_json, output_format, query)?;
-                print_formatted_output(data, output_format)?;
-                println!();
-            }
-            Err(e) => {
-                eprintln!("Error fetching stats: {}", e);
+    println!("Streaming node {} stats (Ctrl+C to stop)...\n", node_id);
+
+    loop {
+        tokio::select! {
+            _ = signal::ctrl_c() => {
+                println!("\nStopping stats stream...");
                 break;
             }
+            result = stream.next() => {
+                match result {
+                    Some(Ok(stats)) => {
+                        let stats_json =
+                            serde_json::to_value(stats).context("Failed to serialize stats")?;
+                        let data = handle_output(stats_json, output_format, query)?;
+                        print_formatted_output(data, output_format)?;
+                        println!();
+                    }
+                    Some(Err(e)) => {
+                        eprintln!("Error fetching stats: {}", e);
+                        break;
+                    }
+                    None => {
+                        break;
+                    }
+                }
+            }
         }
     }
     Ok(())
@@ -296,19 +329,32 @@ async fn handle_cluster_stats_stream(
     let stats_handler = StatsHandler::new(client);
     let mut stream = stats_handler.stream_cluster(Duration::from_secs(poll_interval));
 
-    while let Some(result) = stream.next().await {
-        match result {
-            Ok(stats) => {
-                let stats_json =
-                    serde_json::to_value(stats).context("Failed to serialize stats")?;
-                let data = handle_output(stats_json, output_format, query)?;
-                print_formatted_output(data, output_format)?;
-                println!();
-            }
-            Err(e) => {
-                eprintln!("Error fetching stats: {}", e);
+    println!("Streaming cluster stats (Ctrl+C to stop)...\n");
+
+    loop {
+        tokio::select! {
+            _ = signal::ctrl_c() => {
+                println!("\nStopping stats stream...");
                 break;
             }
+            result = stream.next() => {
+                match result {
+                    Some(Ok(stats)) => {
+                        let stats_json =
+                            serde_json::to_value(stats).context("Failed to serialize stats")?;
+                        let data = handle_output(stats_json, output_format, query)?;
+                        print_formatted_output(data, output_format)?;
+                        println!();
+                    }
+                    Some(Err(e)) => {
+                        eprintln!("Error fetching stats: {}", e);
+                        break;
+                    }
+                    None => {
+                        break;
+                    }
+                }
+            }
         }
     }
     Ok(())