plugin: remove connection after disconnect (#127)

* change clippy rule

* plugin: remove connection after disconnect

yellowstone-grpc-geyser/src/grpc.rs

@@ -201,6 +201,7 @@ impl<'a> From<&'a ReplicaBlockInfoV2<'a>> for MessageBlockMeta {
 }
 
 #[derive(Debug)]
+#[allow(clippy::large_enum_variant)]
 pub enum Message {
     Slot(MessageSlot),
     Account(MessageAccount),
@@ -274,6 +275,9 @@ enum ClientMessage {
         id: usize,
         filter: Filter,
     },
+    Drop {
+        id: usize,
+    },
 }
 
 #[derive(Debug)]
@@ -388,12 +392,17 @@ impl GrpcService {
                             clients.insert(id, ClientConnection { filter, stream_tx });
                             CONNECTIONS_TOTAL.inc();
                         }
-                        ClientMessage::Update {id,filter} => {
+                        ClientMessage::Update { id, filter } => {
                             if let Some(client) = clients.get_mut(&id) {
                                 info!("{}, update client", id);
                                 client.filter = filter;
                             }
                         }
+                        ClientMessage::Drop { id } => {
+                            if clients.remove(&id).is_some() {
+                                CONNECTIONS_TOTAL.dec();
+                            }
+                        }
                     }
                 }
                 else => break,
@@ -435,6 +444,7 @@ impl Geyser for GrpcService {
         }
 
         let ping_stream_tx = stream_tx.clone();
+        let new_clients_tx = self.new_clients_tx.clone();
         tokio::spawn(async move {
             loop {
                 sleep(Duration::from_secs(10)).await;
@@ -447,6 +457,7 @@ impl Geyser for GrpcService {
                     Err(mpsc::error::TrySendError::Closed(_)) => break,
                 }
             }
+            let _ = new_clients_tx.send(ClientMessage::Drop { id });
         });
 
         let config_filters_limit = self.config.filters.clone();
@@ -475,6 +486,7 @@ impl Geyser for GrpcService {
                     Err(_error) => break,
                 }
             }
+            let _ = new_clients_tx.send(ClientMessage::Drop { id });
         });
 
         Ok(Response::new(ReceiverStream::new(stream_rx)))

yellowstone-grpc-geyser/src/lib.rs

@@ -1,5 +1,3 @@
-#![allow(clippy::large_enum_variant)]
-
 pub mod config;
 pub mod filters;
 pub mod grpc;