fix(rust): leaking connections, clippy and tests + examples

rust/examples/account_sub.rs

@@ -1,5 +1,4 @@
-use helius_laserstream::{subscribe, LaserstreamConfig};
-use yellowstone_grpc_proto::geyser::{SubscribeRequest, SubscribeRequestFilterAccounts};
+use helius_laserstream::{subscribe, LaserstreamConfig, grpc::{SubscribeRequest, SubscribeRequestFilterAccounts}};
 use futures::StreamExt;
 use std::env;
 
@@ -32,10 +31,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     while let Some(result) = stream.next().await {
         match result {
             Ok(update) => {
-                println!("{:?}", update);
+                println!("{update:?}");
             }
             Err(e) => {
-                eprintln!("Error: {:?}", e);
+                eprintln!("Error: {e:?}");
                 break;
             }
         }

rust/examples/accounts_data_slice_sub.rs

@@ -1,7 +1,6 @@
-use helius_laserstream::{subscribe, LaserstreamConfig};
-use yellowstone_grpc_proto::geyser::{
+use helius_laserstream::{subscribe, LaserstreamConfig, grpc::{
     SubscribeRequest, SubscribeRequestFilterAccounts, SubscribeRequestAccountsDataSlice
-};
+}};
 use futures::StreamExt;
 use std::env;
 
@@ -41,10 +40,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     while let Some(result) = stream.next().await {
         match result {
             Ok(update) => {
-                println!("{:?}", update);
+                println!("{update:?}");
             }
             Err(e) => {
-                eprintln!("Error: {:?}", e);
+                eprintln!("Error: {e:?}");
                 break;
             }
         }

rust/examples/basic_usage.rs

@@ -48,10 +48,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     while let Some(result) = stream.next().await {
         match result {
             Ok(update) => {
-                println!("{:?}", update);
+                println!("{update:?}");
             }
             Err(e) => {
-                eprintln!("Stream error: {}", e);
+                eprintln!("Stream error: {e}");
             }
         }
     }

rust/examples/block_meta_sub.rs

@@ -1,5 +1,4 @@
-use helius_laserstream::{subscribe, LaserstreamConfig};
-use yellowstone_grpc_proto::geyser::{SubscribeRequest, SubscribeRequestFilterBlocksMeta};
+use helius_laserstream::{subscribe, LaserstreamConfig, grpc::{SubscribeRequest, SubscribeRequestFilterBlocksMeta}};
 use futures::StreamExt;
 use std::env;
 
@@ -29,14 +28,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     while let Some(result) = stream.next().await {
         match result {
             Ok(update) => {
-                println!("{:?}", update);
+                println!("{update:?}");
                 count += 1;
                 if count >= 10 {
                     break;
                 }
             }
             Err(e) => {
-                eprintln!("Error: {:?}", e);
+                eprintln!("Error: {e:?}");
                 break;
             }
         }

rust/examples/block_sub.rs

@@ -1,5 +1,4 @@
-use helius_laserstream::{subscribe, LaserstreamConfig};
-use yellowstone_grpc_proto::geyser::{SubscribeRequest, SubscribeRequestFilterBlocks};
+use helius_laserstream::{subscribe, LaserstreamConfig, grpc::{SubscribeRequest, SubscribeRequestFilterBlocks}};
 use futures::StreamExt;
 use std::env;
 
@@ -33,10 +32,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     while let Some(result) = stream.next().await {
         match result {
             Ok(update) => {
-                println!("{:?}", update);
+                println!("{update:?}");
             }
             Err(e) => {
-                eprintln!("Error: {:?}", e);
+                eprintln!("Error: {e:?}");
                 break;
             }
         }

rust/examples/channel-options-example.rs

@@ -63,10 +63,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     while let Some(update) = stream.next().await {
         match update {
             Ok(update) => {
-                println!("Received update: {:?}", update);
+                println!("Received update: {update:?}");
             }
             Err(e) => {
-                eprintln!("Stream error: {:?}", e);
+                eprintln!("Stream error: {e:?}");
                 break;
             }
         }

rust/examples/compression-advanced.rs

@@ -14,21 +14,17 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .expect("Endpoint not set");
 
     // Example 1: Manual compression configuration
-    let mut channel_options = ChannelOptions::default();
-
-    // Send using zstd compression
-    channel_options.send_compression = Some(CompressionEncoding::Zstd);
-
-    // Accept both compression types (order matters - first is preferred)
-    channel_options.accept_compression = Some(vec![
-        CompressionEncoding::Zstd,  // Prefer zstd
-        CompressionEncoding::Gzip,  // Fallback to gzip
-    ]);
-
-    // Other performance options
-    channel_options.max_decoding_message_size = Some(2_000_000_000); // 2GB
-    channel_options.http2_keep_alive_interval_secs = Some(20);
-    channel_options.tcp_nodelay = Some(true);
+    let channel_options = ChannelOptions {
+        send_compression: Some(CompressionEncoding::Zstd),
+        accept_compression: Some(vec![
+            CompressionEncoding::Zstd,  // Prefer zstd
+            CompressionEncoding::Gzip,  // Fallback to gzip
+        ]),
+        max_decoding_message_size: Some(2_000_000_000), // 2GB
+        http2_keep_alive_interval_secs: Some(20),
+        tcp_nodelay: Some(true),
+        ..Default::default()
+    };
 
     let config = LaserstreamConfig::new(endpoint.clone(), api_key.clone())
         .with_channel_options(channel_options);
@@ -56,7 +52,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 }
             }
             Err(e) => {
-                eprintln!("Error: {:?}", e);
+                eprintln!("Error: {e:?}");
                 break;
             }
         }
@@ -87,7 +83,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 }
             }
             Err(e) => {
-                eprintln!("Error: {:?}", e);
+                eprintln!("Error: {e:?}");
                 break;
             }
         }

rust/examples/compression-example.rs

@@ -48,12 +48,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 }
             }
             Err(e) => {
-                eprintln!("Error: {:?}", e);
+                eprintln!("Error: {e:?}");
                 break;
             }
         }
     }
 
-    println!("Received {} compressed slot updates", count);
+    println!("Received {count} compressed slot updates");
     Ok(())
 }

rust/examples/entry_sub.rs

@@ -1,5 +1,4 @@
-use helius_laserstream::{subscribe, LaserstreamConfig};
-use yellowstone_grpc_proto::geyser::{SubscribeRequest, SubscribeRequestFilterEntry};
+use helius_laserstream::{subscribe, LaserstreamConfig, grpc::{SubscribeRequest, SubscribeRequestFilterEntry}};
 use futures::StreamExt;
 use std::env;
 
@@ -28,10 +27,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     while let Some(result) = stream.next().await {
         match result {
             Ok(update) => {
-                println!("{:?}", update);
+                println!("{update:?}");
             }
             Err(e) => {
-                eprintln!("Error: {:?}", e);
+                eprintln!("Error: {e:?}");
                 break;
             }
         }

rust/examples/preprocessed_transaction_sub.rs

@@ -44,10 +44,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         match result {
             Ok(update) => {
                 // Print the raw debug output
-                println!("{:#?}", update);
+                println!("{update:#?}");
             }
             Err(e) => {
-                eprintln!("Stream error: {:?}", e);
+                eprintln!("Stream error: {e:?}");
                 break;
             }
         }

rust/examples/slot_sub.rs

@@ -1,5 +1,4 @@
-use helius_laserstream::{subscribe, LaserstreamConfig};
-use yellowstone_grpc_proto::geyser::{SubscribeRequest, SubscribeRequestFilterSlots};
+use helius_laserstream::{subscribe, LaserstreamConfig, grpc::{SubscribeRequest, SubscribeRequestFilterSlots}};
 use futures::StreamExt;
 use std::env;
 
@@ -35,10 +34,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     while let Some(result) = stream.next().await {
         match result {
             Ok(update) => {
-                println!("{:?}", update);
+                println!("{update:?}");
             }
             Err(e) => {
-                eprintln!("Error: {:?}", e);
+                eprintln!("Error: {e:?}");
                 break;
             }
         }

rust/examples/stream_write_example.rs

@@ -46,7 +46,6 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let mut stream = Box::pin(stream);
 
     let message_count = Arc::new(AtomicU32::new(0));
-    let handle_clone = handle.clone();
     let count_clone = message_count.clone();
 
     // Spawn a task to add subscriptions dynamically
@@ -73,8 +72,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             ..Default::default()
         };
 
-        if let Err(e) = handle_clone.write(transaction_request).await {
-            eprintln!("❌ Failed to add transaction subscription: {}", e);
+        if let Err(e) = handle.write(transaction_request).await {
+            eprintln!("❌ Failed to add transaction subscription: {e}");
         } else {
             println!("✅ Successfully added transaction subscription");
         }
@@ -102,8 +101,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             ..Default::default()
         };
 
-        if let Err(e) = handle_clone.write(block_request).await {
-            eprintln!("❌ Failed to add block subscription: {}", e);
+        if let Err(e) = handle.write(block_request).await {
+            eprintln!("❌ Failed to add block subscription: {e}");
         } else {
             println!("✅ Successfully added block subscription");
         }
@@ -138,7 +137,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 }
             }
             Err(e) => {
-                eprintln!("❌ Stream error: {}", e);
+                eprintln!("❌ Stream error: {e}");
                 // The stream will automatically reconnect
             }
         }

rust/examples/transaction_status_sub.rs

@@ -1,5 +1,4 @@
-use helius_laserstream::{subscribe, LaserstreamConfig};
-use yellowstone_grpc_proto::geyser::{SubscribeRequest, SubscribeRequestFilterTransactions};
+use helius_laserstream::{subscribe, LaserstreamConfig, grpc::{SubscribeRequest, SubscribeRequestFilterTransactions}};
 use futures::StreamExt;
 use std::env;
 
@@ -32,14 +31,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     while let Some(result) = stream.next().await {
         match result {
             Ok(update) => {
-                println!("{:?}", update);
+                println!("{update:?}");
                 count += 1;
                 if count >= 10 {
                     break;
                 }
             }
             Err(e) => {
-                eprintln!("Error: {:?}", e);
+                eprintln!("Error: {e:?}");
                 break;
             }
         }

rust/examples/transaction_sub.rs

@@ -1,5 +1,4 @@
-use helius_laserstream::{subscribe, LaserstreamConfig};
-use yellowstone_grpc_proto::geyser::{SubscribeRequest, SubscribeRequestFilterTransactions};
+use helius_laserstream::{subscribe, LaserstreamConfig, grpc::{SubscribeRequest, SubscribeRequestFilterTransactions}};
 use futures::StreamExt;
 use std::env;
 
@@ -33,14 +32,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     while let Some(result) = stream.next().await {
         match result {
             Ok(update) => {
-                println!("{:?}", update);
+                println!("{update:?}");
                 count += 1;
                 if count >= 5 {
                     break;
                 }
             }
             Err(e) => {
-                eprintln!("Error: {:?}", e);
+                eprintln!("Error: {e:?}");
                 break;
             }
         }

rust/examples/verify_no_internal_filters.rs

@@ -36,7 +36,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                         }
                     }
                     Err(e) => {
-                        eprintln!("Error: {:?}", e);
+                        eprintln!("Error: {e:?}");
                         break;
                     }
                 }
@@ -47,7 +47,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     match timeout {
         Ok(_) => {
             if count > 0 {
-                println!("WARNING: Received {} updates when none were expected!", count);
+                println!("WARNING: Received {count} updates when none were expected!");
             }
         }
         Err(_) => {
@@ -88,7 +88,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 }
             }
             Some(Err(e)) => {
-                eprintln!("Stream error: {:?}", e);
+                eprintln!("Stream error: {e:?}");
                 break;
             }
             None => {
@@ -98,7 +98,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         }
     }
 
-    println!("\n✓ Successfully verified {} slot updates contain no internal filters", verified_count);
+    println!("\n✓ Successfully verified {verified_count} slot updates contain no internal filters");
 
     Ok(())
 }