send request with subscribe

helius-labs · Dec 4, 2023 · a3bff42 · a3bff42
1 parent 7bd3596
commit a3bff42
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 6 deletions.
diff --git a/examples/rust/src/bin/client.rs b/examples/rust/src/bin/client.rs
@@ -524,11 +524,7 @@ async fn geyser_subscribe(
     request: SubscribeRequest,
     resub: usize,
 ) -> anyhow::Result<()> {
-    let (mut subscribe_tx, mut stream) = client.subscribe().await?;
-    subscribe_tx
-        .send(request)
-        .await
-        .map_err(GeyserGrpcClientError::SubscribeSendError)?;
+    let (mut subscribe_tx, mut stream) = client.subscribe_with_request(Some(request)).await?;
 
     info!("stream opened");
     let mut counter = 0;

diff --git a/yellowstone-grpc-client/src/lib.rs b/yellowstone-grpc-client/src/lib.rs
@@ -1,3 +1,5 @@
+use http::request;
+
 use {
     bytes::Bytes,
     futures::{
@@ -173,7 +175,23 @@ impl<F: Interceptor> GeyserGrpcClient<F> {
         impl Sink<SubscribeRequest, Error = mpsc::SendError>,
         impl Stream<Item = Result<SubscribeUpdate, Status>>,
     )> {
-        let (subscribe_tx, subscribe_rx) = mpsc::unbounded();
+        self.subscribe_with_request(None).await
+    }
+
+    pub async fn subscribe_with_request(
+        &mut self,
+        request: Option<SubscribeRequest>,
+    ) -> GeyserGrpcClientResult<(
+        impl Sink<SubscribeRequest, Error = mpsc::SendError>,
+        impl Stream<Item = Result<SubscribeUpdate, Status>>,
+    )> {
+        let (mut subscribe_tx, subscribe_rx) = mpsc::unbounded();
+        if let Some(request) = request {
+            subscribe_tx
+                .send(request)
+                .await
+                .map_err(GeyserGrpcClientError::SubscribeSendError)?;
+        }
         let response: Response<Streaming<SubscribeUpdate>> =
             self.geyser.subscribe(subscribe_rx).await?;
         Ok((subscribe_tx, response.into_inner()))