websocket pubsub

src/bridge.rs

@@ -10,7 +10,7 @@ use std::{ops::Deref, str::FromStr, sync::Arc};
 use anyhow::bail;
 use reqwest::Url;
 
-use jsonrpsee::server::ServerBuilder;
+use jsonrpsee::{server::ServerBuilder, types::SubscriptionResult, SubscriptionSink};
 use solana_client::{
     nonblocking::{rpc_client::RpcClient, tpu_client::TpuClient},
     rpc_config::{RpcContextConfig, RpcRequestAirdropConfig},
@@ -211,6 +211,17 @@ impl LiteRpcServer for LiteBridge {
             .unwrap()
             .to_string())
     }
+
+    fn signature_subscribe(
+        &self,
+        sink: SubscriptionSink,
+        signature: String,
+        commitment_config: CommitmentConfig,
+    ) -> SubscriptionResult {
+        self.get_block_listner(commitment_config)
+            .signature_subscribe(signature, sink);
+        Ok(())
+    }
 }
 
 impl Deref for LiteBridge {

src/configs.rs

@@ -4,12 +4,12 @@ use serde::{Deserialize, Serialize};
 #[derive(Debug, Default, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct SendTransactionConfig {
-//    #[serde(default)]
+    //    #[serde(default)]
-//    pub skip_preflight: bool,
+    //    pub skip_preflight: bool,
-//    #[serde(default)]
+    //    #[serde(default)]
-//    pub preflight_commitment: CommitmentLevel,
+    //    pub preflight_commitment: CommitmentLevel,
     #[serde(default)]
     pub encoding: BinaryEncoding,
     pub max_retries: Option<u16>,
-//    pub min_context_slot: Option<Slot>,
+    //    pub min_context_slot: Option<Slot>,
 }

src/rpc.rs

@@ -4,6 +4,7 @@ use solana_client::rpc_config::{
     RpcContextConfig, RpcRequestAirdropConfig, RpcSignatureStatusConfig,
 };
 use solana_client::rpc_response::{Response as RpcResponse, RpcBlockhash, RpcVersionInfo};
+use solana_sdk::commitment_config::CommitmentConfig;
 use solana_transaction_status::TransactionStatus;
 
 use crate::configs::SendTransactionConfig;
@@ -42,4 +43,7 @@ pub trait LiteRpc {
         lamports: u64,
         config: Option<RpcRequestAirdropConfig>,
     ) -> Result<String>;
+
+    #[subscription(name = "signatureSubscribe", unsubscribe="signatureUnsubscribe", item=Option<TransactionError>)]
+    fn signature_subscribe(&self, signature: String, commitment_config: CommitmentConfig);
 }

src/workers/block_listenser.rs

@@ -4,12 +4,14 @@ use std::sync::Arc;
 use anyhow::{bail, Context};
 use dashmap::DashMap;
 use futures::StreamExt;
+use jsonrpsee::SubscriptionSink;
 use log::info;
 use solana_client::nonblocking::pubsub_client::PubsubClient;
 use solana_client::nonblocking::rpc_client::RpcClient;
 use solana_client::rpc_config::{RpcBlockSubscribeConfig, RpcBlockSubscribeFilter};
 use solana_sdk::commitment_config::{CommitmentConfig, CommitmentLevel};
 
+use solana_sdk::transaction::TransactionError;
 use solana_transaction_status::{TransactionConfirmationStatus, TransactionStatus};
 use tokio::sync::RwLock;
 use tokio::task::JoinHandle;
@@ -24,6 +26,7 @@ pub struct BlockListener {
     latest_block_hash: Arc<RwLock<String>>,
     block_height: Arc<AtomicU64>,
     commitment_config: CommitmentConfig,
+    signature_subscribers: Arc<DashMap<String, SubscriptionSink>>,
 }
 
 impl BlockListener {
@@ -44,6 +47,7 @@ impl BlockListener {
             latest_block_hash: Arc::new(RwLock::new(latest_block_hash.to_string())),
             block_height: Arc::new(AtomicU64::new(block_height)),
             commitment_config,
+            signature_subscribers: Default::default(),
         })
     }
 
@@ -70,6 +74,14 @@ impl BlockListener {
         )
     }
 
+    pub fn signature_subscribe(&self, signature: String, sink: SubscriptionSink) {
+        self.signature_subscribers.insert(signature, sink).unwrap();
+    }
+
+    pub fn signature_un_subscribe(&self, signature: String) {
+        self.signature_subscribers.remove(&signature);
+    }
+
     pub fn listen(self) -> JoinHandle<anyhow::Result<()>> {
         tokio::spawn(async move {
             info!("Subscribing to blocks");
@@ -124,12 +136,18 @@ impl BlockListener {
                 for sig in signatures {
                     info!("{comfirmation_status:?} {sig}");
 
+                    // subscribers
+                    if let Some((_sig, mut sink)) = self.signature_subscribers.remove(&sig) {
+                        // none if transaction succeeded
+                        sink.send::<Option<TransactionError>>(&None).unwrap();
+                    }
+
                     self.blocks.insert(
                         sig,
                         TransactionStatus {
                             slot,
                             confirmations: None, //TODO: talk about this
-                            status: Ok(()),         // legacy field
+                            status: Ok(()),      // legacy field
                             err: None,
                             confirmation_status: Some(comfirmation_status.clone()),
                         },