Add gossip notifications / cluster node notifications to geyser

Cargo.toml

@@ -16,14 +16,19 @@ repository = "https://github.com/rpcpool/yellowstone-grpc"
 license = "Apache-2.0"
 keywords = ["solana"]
 
+
 [workspace.dependencies]
-solana-account-decoder = "=1.17.5"
-solana-geyser-plugin-interface = "=1.17.5"
-solana-logger = "=1.17.5"
-solana-sdk = "=1.17.5"
-solana-transaction-status = "=1.17.5"
+solana-account-decoder = { path = "../solana-geyzer-cluster/account-decoder" }
+solana-geyser-plugin-interface = { path = "../solana-geyzer-cluster/geyser-plugin-interface" }
+solana-logger = { path = "../solana-geyzer-cluster/logger" }
+solana-sdk = { path = "../solana-geyzer-cluster/sdk" }
+solana-transaction-status = { path = "../solana-geyzer-cluster/transaction-status" }
 
 [profile.release]
 debug = true
 lto = true
 codegen-units = 1
+
+[patch.crates-io]
+solana-program = { path = "../solana-geyzer-cluster/sdk/program" }
+solana-zk-token-sdk = { path = "../solana-geyzer-cluster/zk-token-sdk" }

examples/rust/src/bin/client.rs

@@ -323,6 +323,7 @@ impl Action {
                         commitment: commitment.map(|x| x as i32),
                         accounts_data_slice,
                         ping,
+                        cluster_info: vec![],
                     },
                     args.resub.unwrap_or(0),
                 ))
@@ -580,6 +581,7 @@ async fn geyser_subscribe(
                     commitment: None,
                     accounts_data_slice: Vec::default(),
                     ping: None,
+                    cluster_info: vec![],
                 })
                 .await
                 .map_err(GeyserGrpcClientError::SubscribeSendError)?;

examples/rust/src/bin/tx-blocktime.rs

@@ -105,6 +105,7 @@ async fn main() -> anyhow::Result<()> {
             commitment: Some(commitment as i32),
             accounts_data_slice: vec![],
             ping: None,
+            cluster_info: vec![],
         })
         .await?;
 

yellowstone-grpc-client/src/lib.rs

@@ -1,3 +1,4 @@
+use std::collections::HashSet;
 use {
     bytes::Bytes,
     futures::{
@@ -191,6 +192,7 @@ impl<F: Interceptor> GeyserGrpcClient<F> {
         commitment: Option<CommitmentLevel>,
         accounts_data_slice: Vec<SubscribeRequestAccountsDataSlice>,
         ping: Option<SubscribeRequestPing>,
+        cluster_info: Option<HashSet<String>>,
     ) -> GeyserGrpcClientResult<impl Stream<Item = Result<SubscribeUpdate, Status>>> {
         self.subscribe_once2(SubscribeRequest {
             slots,
@@ -202,6 +204,7 @@ impl<F: Interceptor> GeyserGrpcClient<F> {
             commitment: commitment.map(|value| value as i32),
             accounts_data_slice,
             ping,
+            cluster_info: cluster_info.unwrap_or(HashSet::new()).into_iter().collect(),
         })
         .await
     }

yellowstone-grpc-geyser/src/filters.rs

@@ -6,8 +6,8 @@ use {
             ConfigGrpcFiltersTransactions,
         },
         grpc::{
-            Message, MessageAccount, MessageBlock, MessageBlockMeta, MessageEntry, MessageRef,
-            MessageSlot, MessageTransaction,
+            Message, MessageAccount, MessageBlock, MessageBlockMeta, MessageClusterInfo,
+            MessageEntry, MessageRef, MessageSlot, MessageTransaction,
         },
     },
     base64::{engine::general_purpose::STANDARD as base64_engine, Engine},
@@ -31,6 +31,7 @@ use {
 
 #[derive(Debug, Clone)]
 pub struct Filter {
+    cluster_info: FilterClusterInfo,
     accounts: FilterAccounts,
     slots: FilterSlots,
     transactions: FilterTransactions,
@@ -45,6 +46,7 @@ pub struct Filter {
 impl Filter {
     pub fn new(config: &SubscribeRequest, limit: &ConfigGrpcFilters) -> anyhow::Result<Self> {
         Ok(Self {
+            cluster_info: FilterClusterInfo::new(&config.cluster_info)?, //no config for now
             accounts: FilterAccounts::new(&config.accounts, &limit.accounts)?,
             slots: FilterSlots::new(&config.slots, &limit.slots)?,
             transactions: FilterTransactions::new(&config.transactions, &limit.transactions)?,
@@ -96,6 +98,7 @@ impl Filter {
             Message::Entry(message) => self.entry.get_filters(message),
             Message::Block(message) => self.blocks.get_filters(message),
             Message::BlockMeta(message) => self.blocks_meta.get_filters(message),
+            Message::ClusterInfo(message) => self.cluster_info.get_filters(message),
         }
     }
 
@@ -127,6 +130,31 @@ impl Filter {
     }
 }
 
+#[derive(Debug, Default, Clone)]
+struct FilterClusterInfo {
+    filter_cluster_info: Vec<String>,
+}
+
+impl FilterClusterInfo {
+    fn new(configs: &[String]) -> anyhow::Result<Self> {
+        Ok(Self {
+            filter_cluster_info: configs.iter().map(|name| name.to_string()).collect(),
+        })
+    }
+
+    fn get_filters<'a>(
+        &self,
+        message: &'a MessageClusterInfo,
+    ) -> Vec<(Vec<String>, MessageRef<'a>)> {
+        let subscriptions = self
+            .filter_cluster_info
+            .iter()
+            .map(|name| name.to_string())
+            .collect();
+        vec![(subscriptions, MessageRef::ClusterInfo(message))]
+    }
+}
+
 #[derive(Debug, Default, Clone)]
 struct FilterAccounts {
     filters: Vec<(String, FilterAccountsData)>,
@@ -846,6 +874,7 @@ mod tests {
             commitment: None,
             accounts_data_slice: Vec::new(),
             ping: None,
+            cluster_info: vec![],
         };
         let limit = ConfigGrpcFilters::default();
         let filter = Filter::new(&config, &limit);
@@ -875,6 +904,7 @@ mod tests {
             commitment: None,
             accounts_data_slice: Vec::new(),
             ping: None,
+            cluster_info: vec![],
         };
         let mut limit = ConfigGrpcFilters::default();
         limit.accounts.any = false;
@@ -909,6 +939,7 @@ mod tests {
             commitment: None,
             accounts_data_slice: Vec::new(),
             ping: None,
+            cluster_info: vec![],
         };
         let mut limit = ConfigGrpcFilters::default();
         limit.transactions.any = false;
@@ -942,6 +973,7 @@ mod tests {
             commitment: None,
             accounts_data_slice: Vec::new(),
             ping: None,
+            cluster_info: vec![],
         };
         let mut limit = ConfigGrpcFilters::default();
         limit.transactions.any = false;
@@ -981,6 +1013,7 @@ mod tests {
             commitment: None,
             accounts_data_slice: Vec::new(),
             ping: None,
+            cluster_info: vec![],
         };
         let limit = ConfigGrpcFilters::default();
         let filter = Filter::new(&config, &limit).unwrap();
@@ -1024,6 +1057,7 @@ mod tests {
             commitment: None,
             accounts_data_slice: Vec::new(),
             ping: None,
+            cluster_info: vec![],
         };
         let limit = ConfigGrpcFilters::default();
         let filter = Filter::new(&config, &limit).unwrap();
@@ -1067,6 +1101,7 @@ mod tests {
             commitment: None,
             accounts_data_slice: Vec::new(),
             ping: None,
+            cluster_info: vec![],
         };
         let limit = ConfigGrpcFilters::default();
         let filter = Filter::new(&config, &limit).unwrap();
@@ -1116,6 +1151,7 @@ mod tests {
             commitment: None,
             accounts_data_slice: Vec::new(),
             ping: None,
+            cluster_info: vec![],
         };
         let limit = ConfigGrpcFilters::default();
         let filter = Filter::new(&config, &limit).unwrap();
@@ -1167,6 +1203,7 @@ mod tests {
             commitment: None,
             accounts_data_slice: Vec::new(),
             ping: None,
+            cluster_info: vec![],
         };
         let limit = ConfigGrpcFilters::default();
         let filter = Filter::new(&config, &limit).unwrap();