removing dependency of jsonrpsee for core and services

Cargo.lock

@@ -2299,6 +2299,7 @@ version = "0.2.0"
 dependencies = [
  "anyhow",
  "async-channel",
+ "async-trait",
  "base64 0.21.0",
  "bench",
  "bincode",
@@ -3962,6 +3963,7 @@ name = "solana-lite-rpc-core"
 version = "0.2.0"
 dependencies = [
  "anyhow",
+ "async-trait",
  "base64 0.21.0",
  "bincode",
  "bs58",
@@ -3969,7 +3971,6 @@ dependencies = [
  "chrono",
  "dashmap",
  "futures",
- "jsonrpsee",
  "log",
  "quinn",
  "rustls 0.20.8",
@@ -4001,7 +4002,6 @@ dependencies = [
  "chrono",
  "dashmap",
  "futures",
- "jsonrpsee",
  "lazy_static",
  "log",
  "prometheus",

Cargo.toml

@@ -48,4 +48,5 @@ async-channel = "1.8.0"
 quinn = "0.9.3"
 rustls = { version = "=0.20.8", default-features = false }
 solana-lite-rpc-services = {path = "services", version="0.2.0"}
-solana-lite-rpc-core = {path = "core", version="0.2.0"}
+solana-lite-rpc-core = {path = "core", version="0.2.0"}
+async-trait = "0.1.68"

core/Cargo.toml

@@ -29,7 +29,7 @@ bytes = { workspace = true }
 anyhow = { workspace = true }
 log = { workspace = true }
 dashmap = { workspace = true }
-jsonrpsee = { workspace = true }
 quinn = { workspace = true }
 chrono = { workspace = true }
-rustls = { workspace = true }
+rustls = { workspace = true }
+async-trait = { workspace = true }

core/src/lib.rs

@@ -7,5 +7,5 @@ pub mod rotating_queue;
 pub mod solana_utils;
 pub mod structures;
 pub mod subscription_handler;
-
+pub mod subscription_sink;
 pub type AnyhowJoinHandle = tokio::task::JoinHandle<anyhow::Result<()>>;

core/src/subscription_handler.rs

@@ -1,20 +1,21 @@
 use std::{sync::Arc, time::Duration};
 
 use dashmap::DashMap;
-use jsonrpsee::{SubscriptionMessage, SubscriptionSink};
-use solana_rpc_client_api::response::{Response as RpcResponse, RpcResponseContext};
 use solana_sdk::{
     commitment_config::{CommitmentConfig, CommitmentLevel},
     slot_history::Slot,
 };
 use tokio::time::Instant;
 
-use crate::block_processor::TransactionInfo;
+use crate::{block_processor::TransactionInfo, subscription_sink::SubscriptionSink};
+
+
+pub type SubscptionHanderSink = Arc<dyn SubscriptionSink + Sync + Send>;
 
 #[derive(Clone, Default)]
 pub struct SubscriptionHandler {
     pub signature_subscribers:
-        Arc<DashMap<(String, CommitmentConfig), (SubscriptionSink, Instant)>>,
+        Arc<DashMap<(String, CommitmentConfig), (SubscptionHanderSink, Instant)>>,
 }
 
 impl SubscriptionHandler {
@@ -38,7 +39,7 @@ impl SubscriptionHandler {
         &self,
         signature: String,
         commitment_config: CommitmentConfig,
-        sink: SubscriptionSink,
+        sink: SubscptionHanderSink,
     ) {
         let commitment_config = Self::get_supported_commitment_config(commitment_config);
         self.signature_subscribers
@@ -62,18 +63,11 @@ impl SubscriptionHandler {
             .remove(&(transaction_info.signature.clone(), commitment_config))
         {
             // none if transaction succeeded
-            let _res = sink
+            sink
                 .send(
-                    SubscriptionMessage::from_json(&RpcResponse {
-                        context: RpcResponseContext {
-                            slot,
-                            api_version: None,
-                        },
-                        value: serde_json::json!({ "err": transaction_info.err }),
-                    })
-                    .unwrap(),
-                )
-                .await;
+                    slot,
+                    serde_json::json!({ "err": transaction_info.err })
+                ).await;
         }
     }
 

lite-rpc/Cargo.toml

@@ -39,7 +39,7 @@ async-channel = { workspace = true }
 quinn = { workspace = true }
 solana-lite-rpc-core = { workspace = true }
 solana-lite-rpc-services = { workspace = true }
-
+async-trait = { workspace = true }
 tokio = { version = "1.28.2", features = ["full", "fs"]}
 tokio-postgres = { version = "0.7.8", features = ["with-chrono-0_4"] }
 

lite-rpc/src/bridge.rs

@@ -3,7 +3,7 @@ use crate::{
     encoding::BinaryEncoding,
     postgres::Postgres,
     rpc::LiteRpcServer,
-    DEFAULT_MAX_NUMBER_OF_TXS_IN_QUEUE,
+    DEFAULT_MAX_NUMBER_OF_TXS_IN_QUEUE, jsonrpsee_subscrption_handler_sink::JsonRpseeSubscriptionHandlerSink,
 };
 
 use solana_lite_rpc_services::{
@@ -507,8 +507,9 @@ impl LiteRpcServer for LiteBridge {
         RPC_SIGNATURE_SUBSCRIBE.inc();
         let sink = pending.accept().await?;
 
+        let jsonrpsee_sink = JsonRpseeSubscriptionHandlerSink::new(sink);
         self.block_listner
-            .signature_subscribe(signature, commitment_config, sink);
+            .signature_subscribe(signature, commitment_config, Arc::new(jsonrpsee_sink));
 
         Ok(())
     }

lite-rpc/src/lib.rs

@@ -8,6 +8,7 @@ pub mod encoding;
 pub mod errors;
 pub mod postgres;
 pub mod rpc;
+pub mod jsonrpsee_subscrption_handler_sink;
 
 #[from_env]
 pub const DEFAULT_RPC_ADDR: &str = "http://0.0.0.0:8899";

services/Cargo.toml

@@ -29,7 +29,6 @@ bytes = { workspace = true }
 anyhow = { workspace = true }
 log = { workspace = true }
 dashmap = { workspace = true }
-jsonrpsee = { workspace = true }
 prometheus = { workspace = true }
 lazy_static = { workspace = true }
 async-channel = { workspace = true }

services/src/block_listenser.rs

@@ -34,7 +34,7 @@ use solana_lite_rpc_core::{
     notifications::{
         BlockNotification, NotificationMsg, NotificationSender, TransactionUpdateNotification,
     },
-    subscription_handler::SubscriptionHandler,
+    subscription_handler::{SubscriptionHandler, SubscptionHanderSink},
 };
 
 use crate::tx_sender::{TxProps, TxSender};
@@ -109,7 +109,7 @@ impl BlockListener {
         &self,
         signature: String,
         commitment_config: CommitmentConfig,
-        sink: SubscriptionSink,
+        sink: SubscptionHanderSink,
     ) {
         self.subscription_handler
             .signature_subscribe(signature, commitment_config, sink);