From eab62d0599fac7edc1aeb26095ab3dbaf733959b Mon Sep 17 00:00:00 2001
From: Christian Kamm <ckamm@delightful-solutions.de>
Date: Mon, 1 Nov 2021 15:51:05 +0100
Subject: [PATCH] gRPC: send regular ping

The idea was that slot updates would be really frequent and serve as a
ping. But during node startup there can be a long phase of no data.
---
 .../src/accountsdb_plugin_grpc.rs                  | 14 +++++++++++++-
 connector-lib/src/main.rs                          |  3 ++-
 proto/accountsdb.proto                             |  3 +++
 3 files changed, 18 insertions(+), 2 deletions(-)

diff --git a/accountsdb-plugin-grpc/src/accountsdb_plugin_grpc.rs b/accountsdb-plugin-grpc/src/accountsdb_plugin_grpc.rs
index e8679ee..af6f399 100644
--- a/accountsdb-plugin-grpc/src/accountsdb_plugin_grpc.rs
+++ b/accountsdb-plugin-grpc/src/accountsdb_plugin_grpc.rs
@@ -2,7 +2,7 @@ use {
     crate::accounts_selector::AccountsSelector,
     accountsdb_proto::{
         slot_update::Status as SlotUpdateStatus, update::UpdateOneof, AccountWrite, SlotUpdate,
-        SubscribeRequest, Update,
+        SubscribeRequest, Update, Ping,
     },
     bs58,
     futures_util::FutureExt,
@@ -155,6 +155,16 @@ impl AccountsDbPlugin for AccountsDbPluginGrpc {
                 .add_service(server)
                 .serve_with_shutdown(addr, exit_receiver.map(drop)),
         );
+        let sender_c = self.server_broadcast.as_ref().unwrap().clone();
+        rt.spawn(async move {
+            loop {
+                // Don't care about the error if there are no receivers.
+                let _ = sender_c.send(Update {
+                    update_oneof: Some(UpdateOneof::Ping(Ping{})),
+                });
+                tokio::time::sleep(std::time::Duration::from_secs(5)).await;
+            }
+        });
         self.runtime = Some(rt);
 
         Ok(())
@@ -168,6 +178,8 @@ impl AccountsDbPlugin for AccountsDbPluginGrpc {
                 .send(())
                 .expect("sending grpc server termination should succeed");
         }
+
+        // TODO: explicitly shut down runtime?
     }
 
     fn update_account(
diff --git a/connector-lib/src/main.rs b/connector-lib/src/main.rs
index 335baaf..58bb63c 100644
--- a/connector-lib/src/main.rs
+++ b/connector-lib/src/main.rs
@@ -663,7 +663,8 @@ async fn main() {
                         status: "bla".into(),
                     })
                     .unwrap();
-            }
+            },
+            accountsdb_proto::update::UpdateOneof::Ping(_) => {},
         }
     }
 
diff --git a/proto/accountsdb.proto b/proto/accountsdb.proto
index 244ccdd..15e9d07 100644
--- a/proto/accountsdb.proto
+++ b/proto/accountsdb.proto
@@ -17,6 +17,7 @@ message Update {
   oneof update_oneof {
     AccountWrite account_write = 1;
     SlotUpdate slot_update = 2;
+    Ping ping = 3;
   }
 }
 
@@ -43,3 +44,5 @@ message SlotUpdate {
   Status status = 3;
 }
 
+message Ping {
+}