feat(enhanced websocket): Add transactionSubscribe and accountSubscribe

Cargo.toml

@@ -12,18 +12,27 @@ readme = "README.md"
 homepage = "https://www.helius.dev/"
 
 [dependencies]
+base64 = "0.22.1"
+bincode = "1.3.3"
+chrono = { version = "0.4.11", features = ["serde"] }
+futures = "0.3.30"
+futures-util = "0.3.30"
 reqwest = { version = "0.12.3", features = ["json"] }
+semver = "1.0.23"
 serde = "1.0.198"
+serde-enum-str = "0.4.0"
 serde_json = "1.0.116"
+solana-account-decoder = "1.18.12"
+solana-client = "1.18.12"
+solana-program = "1.18.12"
+solana-rpc-client-api = "1.18.12"
 solana-sdk = "1.18.11"
+solana-transaction-status = "1.18.12"
 thiserror = "1.0.58"
 tokio = { version = "1.37.0", features = ["macros", "rt-multi-thread", "net"] }
-chrono = { version = "0.4.11", features = ["serde"] }
-solana-client = "1.18.12"
-solana-program = "1.18.12"
-serde-enum-str = "0.4.0"
-bincode = "1.3.3"
-base64 = "0.22.1"
+tokio-stream = "0.1.15"
+tokio-tungstenite = { version = "0.21.0", features = ["native-tls", "handshake"] }
+url = "2.5.0"
 
 [dev-dependencies]
 mockito = "1.4.0"

README.md

@@ -20,11 +20,11 @@ Remember to run `cargo update` regularly to fetch the latest version of the SDK.
 ### `Helius`
 The SDK provides a [`Helius`](https://github.com/helius-labs/helius-rust-sdk/blob/dev/src/client.rs) instance that can be configured with an API key and a given Solana cluster. Developers can generate a new API key on the [Helius Developer Dashboard](https://dev.helius.xyz/dashboard/app). This instance acts as the main entry point for interacting with the SDK by providing methods to access different Solana and RPC client functionalities. The following code is an example of how to use the SDK to fetch info on [Mad Lad #8420](https://xray.helius.xyz/token/F9Lw3ki3hJ7PF9HQXsBzoY8GyE6sPoEZZdXJBsTTD2rk?network=mainnet):
 ```rust
-use helius::error::HeliusError;
+use helius::error::Result;
 use helius::types::{Cluster, DisplayOptions, GetAssetRequest, GetAssetResponseForAsset};
 
 #[tokio::main]
-async fn main() -> Result<(), HeliusError> {
+async fn main() -> Result<()> {
     let api_key: &str = "YOUR_API_KEY";
     let cluster: Cluster = Cluster::MainnetBeta;
 
@@ -40,7 +40,7 @@ async fn main() -> Result<(), HeliusError> {
         }),
     };
 
-    let response: Result<Option<GetAssetResponseForAsset>, HeliusError> = helius.rpc().get_asset(request).await;
+    let response: Result<Option<GetAssetResponseForAsset>> = helius.rpc().get_asset(request).await;
 
     match response {
         Ok(Some(asset)) => {

examples/enhanced_websocket_accounts.rs

@@ -0,0 +1,24 @@
+use helius::error::Result;
+use helius::types::Cluster;
+use helius::Helius;
+use solana_sdk::pubkey;
+use tokio_stream::StreamExt;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let api_key: &str = "your_api_key";
+    let cluster: Cluster = Cluster::MainnetBeta;
+
+    let helius: Helius = Helius::new_with_ws(api_key, cluster).await.unwrap();
+
+    let key = pubkey!("BtsmiEEvnSuUnKxqXj2PZRYpPJAc7C34mGz8gtJ1DAaH");
+
+    if let Some(ws) = helius.ws() {
+        let (mut stream, _unsub) = ws.account_subscribe(&key, None).await?;
+        while let Some(event) = stream.next().await {
+            println!("{:#?}", event);
+        }
+    }
+
+    Ok(())
+}

examples/enhanced_websocket_transactions.rs

@@ -0,0 +1,29 @@
+use helius::error::Result;
+use helius::types::{Cluster, RpcTransactionsConfig, TransactionSubscribeFilter, TransactionSubscribeOptions};
+use helius::Helius;
+use solana_sdk::pubkey;
+use tokio_stream::StreamExt;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let api_key: &str = "your_api_key";
+    let cluster: Cluster = Cluster::MainnetBeta;
+
+    let helius: Helius = Helius::new_with_ws(api_key, cluster).await.unwrap();
+
+    let key = pubkey!("BtsmiEEvnSuUnKxqXj2PZRYpPJAc7C34mGz8gtJ1DAaH");
+
+    let config = RpcTransactionsConfig {
+        filter: TransactionSubscribeFilter::standard(&key),
+        options: TransactionSubscribeOptions::default(),
+    };
+
+    if let Some(ws) = helius.ws() {
+        let (mut stream, _unsub) = ws.transaction_subscribe(config).await?;
+        while let Some(event) = stream.next().await {
+            println!("{:#?}", event);
+        }
+    }
+
+    Ok(())
+}

examples/get_asset_batch.rs

@@ -1,9 +1,9 @@
-use helius::error::HeliusError;
+use helius::error::Result;
 use helius::types::{Asset, Cluster, GetAssetBatch, GetAssetOptions};
 use helius::Helius;
 
 #[tokio::main]
-async fn main() -> Result<(), HeliusError> {
+async fn main() -> Result<()> {
     let api_key: &str = "your_api_key";
     let cluster: Cluster = Cluster::MainnetBeta;
 
@@ -20,7 +20,7 @@ async fn main() -> Result<(), HeliusError> {
         }),
     };
 
-    let response: Result<Vec<Option<Asset>>, HeliusError> = helius.rpc().get_asset_batch(request).await;
+    let response: Result<Vec<Option<Asset>>> = helius.rpc().get_asset_batch(request).await;
     println!("Assets: {:?}", response);
 
     Ok(())

examples/get_asset_proof_batch.rs

@@ -1,11 +1,11 @@
-use helius::error::HeliusError;
+use helius::error::Result;
 use helius::types::{AssetProof, Cluster, GetAssetProofBatch};
 use helius::Helius;
 
 use std::collections::HashMap;
 
 #[tokio::main]
-async fn main() -> Result<(), HeliusError> {
+async fn main() -> Result<()> {
     let api_key: &str = "your_api_key";
     let cluster: Cluster = Cluster::MainnetBeta;
 
@@ -18,8 +18,7 @@ async fn main() -> Result<(), HeliusError> {
         ],
     };
 
-    let response: Result<HashMap<String, Option<AssetProof>>, HeliusError> =
-        helius.rpc().get_asset_proof_batch(request).await;
+    let response: Result<HashMap<String, Option<AssetProof>>> = helius.rpc().get_asset_proof_batch(request).await;
     println!("Assets: {:?}", response);
 
     Ok(())

examples/get_latest_blockhash.rs

@@ -1,18 +1,18 @@
-use helius::error::HeliusError;
+use helius::error::Result;
 use helius::types::*;
 use helius::Helius;
 
 use solana_client::client_error::ClientError;
 use solana_sdk::hash::Hash;
 
 #[tokio::main]
-async fn main() -> Result<(), HeliusError> {
+async fn main() -> Result<()> {
     let api_key: &str = "your_api_key";
     let cluster: Cluster = Cluster::MainnetBeta;
 
     let helius: Helius = Helius::new(api_key, cluster).unwrap();
 
-    let result: Result<Hash, ClientError> = helius.connection().get_latest_blockhash();
+    let result: std::result::Result<Hash, ClientError> = helius.connection().get_latest_blockhash();
     println!("{:?}", result);
 
     Ok(())

examples/get_parse_transactions.rs

@@ -1,9 +1,9 @@
-use helius::error::HeliusError;
+use helius::error::Result;
 use helius::types::*;
 use helius::Helius;
 
 #[tokio::main]
-async fn main() -> Result<(), HeliusError> {
+async fn main() -> Result<()> {
     let api_key: &str = "your_api_key";
     let cluster: Cluster = Cluster::MainnetBeta;
 
@@ -15,7 +15,7 @@ async fn main() -> Result<(), HeliusError> {
         ],
     };
 
-    let response: Result<Vec<EnhancedTransaction>, HeliusError> = helius.parse_transactions(request).await;
+    let response: Result<Vec<EnhancedTransaction>> = helius.parse_transactions(request).await;
     println!("Assets: {:?}", response);
 
     Ok(())

examples/get_parsed_transaction_history.rs

@@ -1,9 +1,9 @@
-use helius::error::HeliusError;
+use helius::error::Result;
 use helius::types::*;
 use helius::Helius;
 
 #[tokio::main]
-async fn main() -> Result<(), HeliusError> {
+async fn main() -> Result<()> {
     let api_key: &str = "your_api_key";
     let cluster: Cluster = Cluster::MainnetBeta;
 
@@ -14,7 +14,7 @@ async fn main() -> Result<(), HeliusError> {
         before: None,
     };
 
-    let response: Result<Vec<EnhancedTransaction>, HeliusError> = helius.parsed_transaction_history(request).await;
+    let response: Result<Vec<EnhancedTransaction>> = helius.parsed_transaction_history(request).await;
 
     println!("Assets: {:?}", response);
 

examples/get_priority_fee_estimate.rs

@@ -1,9 +1,9 @@
-use helius::error::HeliusError;
+use helius::error::Result;
 use helius::types::*;
 use helius::Helius;
 
 #[tokio::main]
-async fn main() -> Result<(), HeliusError> {
+async fn main() -> Result<()> {
     let api_key: &str = "your_api_key";
     let cluster: Cluster = Cluster::MainnetBeta;
 
@@ -22,8 +22,7 @@ async fn main() -> Result<(), HeliusError> {
         }),
     };
 
-    let response: Result<GetPriorityFeeEstimateResponse, HeliusError> =
-        helius.rpc().get_priority_fee_estimate(request).await;
+    let response: Result<GetPriorityFeeEstimateResponse> = helius.rpc().get_priority_fee_estimate(request).await;
     println!("Assets: {:?}", response);
 
     Ok(())

examples/mint_cnft.rs

@@ -1,10 +1,10 @@
-use helius::error::HeliusError;
+use helius::error::Result;
 use helius::types::*;
 use helius::Helius;
 use serde_json::Value;
 
 #[tokio::main]
-async fn main() -> Result<(), HeliusError> {
+async fn main() -> Result<()> {
     let api_key: &str = "your_api_key";
     let cluster: Cluster = Cluster::MainnetBeta;
 
@@ -39,7 +39,7 @@ async fn main() -> Result<(), HeliusError> {
         confirm_transaction: Some(true),
     };
 
-    let response: Result<MintResponse, HeliusError> = helius.mint_compressed_nft(request).await;
+    let response: Result<MintResponse> = helius.mint_compressed_nft(request).await;
     println!("Assets: {:?}", response);
 
     Ok(())

src/client.rs

@@ -4,6 +4,7 @@ use crate::config::Config;
 use crate::error::Result;
 use crate::rpc_client::RpcClient;
 use crate::types::Cluster;
+use crate::websocket::{EnhancedWebsocket, ENHANCED_WEBSOCKET_URL};
 
 use reqwest::Client;
 use solana_client::rpc_client::RpcClient as SolanaRpcClient;
@@ -19,6 +20,8 @@ pub struct Helius {
     pub client: Client,
     /// A reference-counted RPC client tailored for making requests in a thread-safe manner
     pub rpc_client: Arc<RpcClient>,
+    /// A reference-counted enhanced (geyser) websocket client
+    pub ws_client: Option<Arc<EnhancedWebsocket>>,
 }
 
 impl Helius {
@@ -42,11 +45,33 @@ impl Helius {
         let config: Arc<Config> = Arc::new(Config::new(api_key, cluster)?);
         let client: Client = Client::new();
         let rpc_client: Arc<RpcClient> = Arc::new(RpcClient::new(Arc::new(client.clone()), config.clone())?);
+        Ok(Helius {
+            config,
+            client,
+            rpc_client,
+            ws_client: None,
+        })
+    }
 
+    /// The enhanced websocket is optional, and this method is used to create a new instance of `Helius` with an enhanced websocket client.
+    /// Upon calling this method, the websocket will connect hence the asynchronous function definition omission from the default `new` method.
+    ///
+    /// # Arguments
+    /// * `api_key` - The API key required for authenticating requests made
+    /// * `cluster` - The Solana cluster (Devnet or MainnetBeta) that defines the given network environment
+    /// # Returns
+    /// An instance of `Helius` if successful. A `HeliusError` is returned if an error occurs during configuration or initialization of the HTTP, RPC, or WS client
+    pub async fn new_with_ws(api_key: &str, cluster: Cluster) -> Result<Self> {
+        let config: Arc<Config> = Arc::new(Config::new(api_key, cluster)?);
+        let client: Client = Client::new();
+        let rpc_client: Arc<RpcClient> = Arc::new(RpcClient::new(Arc::new(client.clone()), config.clone())?);
+        let wss: String = format!("{}{}", ENHANCED_WEBSOCKET_URL, api_key);
+        let ws_client: Arc<EnhancedWebsocket> = Arc::new(EnhancedWebsocket::new(&wss).await?);
         Ok(Helius {
             config,
             client,
             rpc_client,
+            ws_client: Some(ws_client),
         })
     }
 
@@ -61,4 +86,8 @@ impl Helius {
     pub fn connection(&self) -> Arc<SolanaRpcClient> {
         self.rpc_client.solana_client.clone()
     }
+
+    pub fn ws(&self) -> Option<Arc<EnhancedWebsocket>> {
+        self.ws_client.clone()
+    }
 }

src/error.rs

@@ -105,6 +105,18 @@ pub enum HeliusError {
     /// This error includes the HTTP status code and message to help with debugging, acting as a generic catch-all for all other errors
     #[error("Unknown error has occurred: HTTP {code} - {text}")]
     Unknown { code: StatusCode, text: String },
+
+    #[error("Unable to connect to server: {0}")]
+    Tungstenite(#[from] tokio_tungstenite::tungstenite::Error),
+
+    #[error("Websocket connection closed (({0})")]
+    WebsocketClosed(String),
+
+    #[error("Enhanced websocket: {message}: {reason}")]
+    EnhancedWebsocket { reason: String, message: String },
+
+    #[error("Url parse error")]
+    UrlParseError(#[from] url::ParseError),
 }
 
 impl HeliusError {

src/factory.rs

@@ -89,6 +89,7 @@ impl HeliusFactory {
             config,
             client,
             rpc_client,
+            ws_client: None,
         })
     }
 }

src/lib.rs

@@ -10,6 +10,7 @@ pub mod rpc_client;
 pub mod types;
 pub mod utils;
 pub mod webhook;
+pub mod websocket;
 
 pub use client::Helius;
 pub use factory::HeliusFactory;