fix all clippy

Cargo.lock

@@ -1249,6 +1249,7 @@ dependencies = [
  "futures",
  "itertools",
  "log",
+ "merge-streams",
  "prometheus",
  "solana-sdk",
  "tokio",
@@ -1717,6 +1718,16 @@ dependencies = [
  "autocfg",
 ]
 
+[[package]]
+name = "merge-streams"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4f84f6452969abd246e7ac1fe4fe75906c76e8ec88d898df9aef37e0f3b6a7c2"
+dependencies = [
+ "futures-core",
+ "pin-project",
+]
+
 [[package]]
 name = "merlin"
 version = "3.0.0"

examples/grpcmultiplex_perfectseq.rs

@@ -97,23 +97,25 @@ pub async fn main() {
             match msg_or_timeout {
                 Ok(Some(offer_block_msg)) => {
                     // collect the offered slots from the channels
-                    if let OfferBlockMsg::NextSlot(label, block_offered) = offer_block_msg {
-                        info!("<< offered slot from {}: {:?}", label, block_offered);
+                    match offer_block_msg {
+                        OfferBlockMsg::NextSlot(label, block_offered) => {
+                            info!("<< offered slot from {}: {:?}", label, block_offered);
 
-                        // TOOD use .parent instead
-                        if block_offered.parent_slot == current_tip {
-                            current_tip = block_offered.slot;
-                            info!("<< take block from {} as new tip {}", label, current_tip);
-                            assert_ne!(current_tip, 0, "must not see uninitialized tip");
+                            // TOOD use .parent instead
+                            if block_offered.parent_slot == current_tip {
+                                current_tip = block_offered.slot;
+                                info!("<< take block from {} as new tip {}", label, current_tip);
+                                assert_ne!(current_tip, 0, "must not see uninitialized tip");
 
-                            emit_block_on_multiplex_output_channel(&sx_multi, current_tip);
-                            tx_tip.send(current_tip).unwrap();
-                            blocks_offered.clear();
-                            continue 'main_loop;
-                        } else {
-                            // keep the block for later
-                            blocks_offered.push(block_offered);
-                            continue 'main_loop;
+                                emit_block_on_multiplex_output_channel(&sx_multi, current_tip);
+                                tx_tip.send(current_tip).unwrap();
+                                blocks_offered.clear();
+                                continue 'main_loop;
+                            } else {
+                                // keep the block for later
+                                blocks_offered.push(block_offered);
+                                continue 'main_loop;
+                            }
                         }
                     }
                     // TODO handle else

src/experimental/grpcmultiplex_fastestwins_channels.rs

@@ -69,7 +69,7 @@ pub async fn create_multiplex(
 
         let mut current_slot: Slot = 0;
 
-        'main_loop: loop {
+        loop {
             let block_cmd = select! {
                 message = &mut merged_futures => {
                     match message {

src/grpcmultiplex_fastestwins.rs

@@ -1,6 +1,6 @@
 use crate::grpc_subscription_autoreconnect::{create_geyser_reconnecting_stream, GrpcSourceConfig};
 use async_stream::stream;
-use futures::{Stream, StreamExt};
+use futures::Stream;
 use itertools::Itertools;
 use log::{debug, info};
 use solana_sdk::clock::Slot;