From 8a9fca6e3babe71158bcf8cb1b9e3381c8fcbe5b Mon Sep 17 00:00:00 2001
From: Kirill Fomichev <fanatid@ya.ru>
Date: Thu, 10 Oct 2024 22:57:16 +0700
Subject: [PATCH] geyser: optimize consuming of new filters (#439)

---
 CHANGELOG.md                        |  1 +
 yellowstone-grpc-geyser/src/grpc.rs | 16 +++++++++++++++-
 2 files changed, 16 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5ef14389..709483e6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -15,6 +15,7 @@ The minor version will be incremented upon a breaking change and the patch versi
 ### Features
 
 - proto: use `gzip`/`zstd` features by default ([#436](https://github.com/rpcpool/yellowstone-grpc/pull/436))
+- geyser: optimize consuming of new filters ([#439](https://github.com/rpcpool/yellowstone-grpc/pull/439))
 
 ### Breaking
 
diff --git a/yellowstone-grpc-geyser/src/grpc.rs b/yellowstone-grpc-geyser/src/grpc.rs
index fd82ebac..ef60bc9d 100644
--- a/yellowstone-grpc-geyser/src/grpc.rs
+++ b/yellowstone-grpc-geyser/src/grpc.rs
@@ -1172,7 +1172,21 @@ impl GrpcService {
         if is_alive {
             'outer: loop {
                 tokio::select! {
-                    message = client_rx.recv() => {
+                    mut message = client_rx.recv() => {
+                        // forward to latest filter
+                        loop {
+                            match client_rx.try_recv() {
+                                Ok(message_new) => {
+                                    message = Some(message_new);
+                                }
+                                Err(mpsc::error::TryRecvError::Empty) => break,
+                                Err(mpsc::error::TryRecvError::Disconnected) => {
+                                    message = None;
+                                    break;
+                                }
+                            }
+                        }
+
                         match message {
                             Some(Some(filter_new)) => {
                                 if let Some(msg) = filter_new.get_pong_msg() {