WIP integrate window patch

Cargo.toml

@@ -9,12 +9,12 @@ authors = ["GroovieGermanikus <[email protected]>"]
 repository = "https://github.com/blockworks-foundation/geyser-grpc-connector"
 
 [dependencies]
-yellowstone-grpc-client = { version = "1.13.0+solana.1.17.15", git = "https://github.com/rpcpool/yellowstone-grpc.git", tag = "v1.12.0+solana.1.17.15" }
-yellowstone-grpc-proto = { version = "1.12.0+solana.1.17.15", git = "https://github.com/rpcpool/yellowstone-grpc.git", tag = "v1.12.0+solana.1.17.15" }
+yellowstone-grpc-client = "1.15.0"
+yellowstone-grpc-proto = "1.14.0"
 
 
 # required for CommitmentConfig
-solana-sdk = "~1.17.15"
+solana-sdk = "~1.17.28"
 
 url = "2.5.0"
 async-stream = "0.3.5"

src/grpc_subscription_autoreconnect_streams.rs

@@ -5,8 +5,9 @@ use log::{debug, info, log, trace, warn, Level};
 use std::time::Duration;
 use tokio::task::JoinHandle;
 use tokio::time::{sleep, timeout};
-use yellowstone_grpc_client::{GeyserGrpcClient, GeyserGrpcClientResult};
+use yellowstone_grpc_client::{GeyserGrpcBuilder, GeyserGrpcBuilderResult, GeyserGrpcClient, GeyserGrpcClientResult, InterceptorXToken};
 use yellowstone_grpc_proto::geyser::{SubscribeRequest, SubscribeUpdate};
+use yellowstone_grpc_proto::tonic::metadata::AsciiMetadataValue;
 use yellowstone_grpc_proto::tonic::Status;
 
 enum ConnectionState<S: Stream<Item = Result<SubscribeUpdate, Status>>> {
@@ -24,41 +25,48 @@ pub fn create_geyser_reconnecting_stream(
 ) -> impl Stream<Item = Message> {
     let mut state = ConnectionState::NotConnected(1);
 
+
+
     // in case of cancellation, we restart from here:
     // thus we want to keep the progression in a state object outside the stream! makro
     let the_stream = stream! {
+
+        let addr = grpc_source.grpc_addr.clone();
+        let grpc_x_token = grpc_source.grpc_x_token.clone();
+        let tls_config = grpc_source.tls_config.clone();
+        // let receive_timeout = grpc_source.timeouts.as_ref().map(|t| t.receive_timeout);
+        let receive_timeout = Some(Duration::from_secs(10)); // FIXME
+
         loop {
             let yield_value;
 
+
             (state, yield_value) = match state {
 
                 ConnectionState::NotConnected(attempt) => {
 
+                    // let connect_timeout = grpc_source.timeouts.as_ref().map(|t| t.connect_timeout);
+                    // let request_timeout = grpc_source.timeouts.as_ref().map(|t| t.request_timeout);
+                    // let subscribe_timeout = grpc_source.timeouts.map(|t| t.subscribe_timeout);
+                    let subscribe_timeout = Some(Duration::from_secs(10)); // FIXME
+
+
+
+
                     let connection_task = tokio::spawn({
-                        let addr = grpc_source.grpc_addr.clone();
-                        let token = grpc_source.grpc_x_token.clone();
-                        let config = grpc_source.tls_config.clone();
-                        let connect_timeout = grpc_source.timeouts.as_ref().map(|t| t.connect_timeout);
-                        let request_timeout = grpc_source.timeouts.as_ref().map(|t| t.request_timeout);
-                        let subscribe_timeout = grpc_source.timeouts.as_ref().map(|t| t.subscribe_timeout);
                         let subscribe_filter = subscribe_filter.clone();
+                        let builder = build_client(grpc_source.clone()).unwrap(); // TODO instead of unwrap, move to Fatal state
                         log!(if attempt > 1 { Level::Warn } else { Level::Debug }, "Connecting attempt #{} to {}", attempt, addr);
                         async move {
 
-                            let connect_result = GeyserGrpcClient::connect_with_timeout(
-                                    addr, token, config,
-                                    connect_timeout,
-                                    request_timeout,
-                                false)
-                                .await;
-                            let mut client = connect_result?;
 
+                            let mut client = builder.connect().await.unwrap(); // FIXME
 
                             debug!("Subscribe with filter {:?}", subscribe_filter);
 
                             let subscribe_result = timeout(subscribe_timeout.unwrap_or(Duration::MAX),
                                 client
-                                    .subscribe_once2(subscribe_filter))
+                                    .subscribe_once(subscribe_filter))
                             .await;
 
                             // maybe not optimal
@@ -88,7 +96,6 @@ pub fn create_geyser_reconnecting_stream(
                 }
 
                 ConnectionState::Ready(mut geyser_stream) => {
-                    let receive_timeout = grpc_source.timeouts.as_ref().map(|t| t.receive_timeout);
                     match timeout(receive_timeout.unwrap_or(Duration::MAX), geyser_stream.next()).await {
                         Ok(Some(Ok(update_message))) => {
                             trace!("> recv update message from {}", grpc_source);
@@ -130,6 +137,34 @@ pub fn create_geyser_reconnecting_stream(
     the_stream
 }
 
+fn build_client(grpc_source_config: GrpcSourceConfig) -> GeyserGrpcBuilderResult<GeyserGrpcBuilder> {
+    let mut builder = GeyserGrpcClient::build_from_shared(grpc_source_config.grpc_addr)?;
+
+    if let Some(tls_config) = grpc_source_config.tls_config {
+        builder = builder.tls_config(tls_config)?;
+    }
+
+    if let Some(timeouts) = grpc_source_config.timeouts {
+
+        builder = builder.timeout(timeouts.connect_timeout);
+
+        builder = builder.timeout(timeouts.request_timeout);
+
+        // subscribe + receive timeout are handled somewhere else
+
+    }
+
+    let x_token: Option<AsciiMetadataValue> = match grpc_source_config.grpc_x_token {
+        Some(x_token) => Some(x_token.try_into()?),
+        None => None,
+    };
+    let interceptor = InterceptorXToken { x_token };
+
+    builder = builder.interceptor(interceptor);
+
+    Ok(builder)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/lib.rs

@@ -3,6 +3,8 @@ use std::collections::HashMap;
 use std::fmt::{Debug, Display};
 use std::time::Duration;
 use yellowstone_grpc_proto::geyser::{CommitmentLevel, SubscribeRequest, SubscribeRequestFilterAccounts, SubscribeRequestFilterBlocks, SubscribeRequestFilterBlocksMeta, SubscribeRequestFilterSlots, SubscribeUpdate};
+use yellowstone_grpc_proto::prost::bytes::Bytes;
+use yellowstone_grpc_proto::tonic;
 use yellowstone_grpc_proto::tonic::transport::ClientTlsConfig;
 
 pub mod channel_plugger;
@@ -31,6 +33,7 @@ pub struct GrpcConnectionTimeouts {
     pub receive_timeout: Duration,
 }
 
+
 #[derive(Clone)]
 pub struct GrpcSourceConfig {
     pub grpc_addr: String,
@@ -78,6 +81,15 @@ impl GrpcSourceConfig {
             timeouts: Some(timeouts),
         }
     }
+
+    pub fn build_tonic_endpoint(&self) -> tonic::transport::Endpoint {
+        let mut endpoint = tonic::transport::Endpoint::from_shared(self.grpc_addr.clone())
+            .expect("grpc_addr must be a valid url");
+        if let Some(tls_config) = &self.tls_config {
+            endpoint = endpoint.tls_config(tls_config.clone()).expect("tls_config must be valid");
+        }
+        endpoint
+    }
 }
 
 #[derive(Clone)]