Add subscriptions to RPC interface

Cargo.toml

@@ -92,8 +92,10 @@ async-trait = "0.1"
 futures = { version = "0.3.1" }
 hex = { version = "0.4.3", default-features = false }
 jsonrpsee = { version = "0.16.2" }
+schnellru = "0.2.1"
 thiserror = { version = "1.0.40" }
 tokio = { version = "1.32.0", default-features = false }
+tokio-stream = "0.1.15"
 tracing = { version = "0.1.37", default-features = false }
 url = "2.2.2"
 
@@ -102,7 +104,6 @@ codegen-units = 1
 inherits = "release"
 lto = true
 
-
 [profile.release]
 opt-level = 3
 panic = "unwind"

client/orchestrator-chain-rpc-interface/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "orchestrator-chain-rpc-interface"
+name = "dc-orchestrator-chain-rpc-interface"
 authors = { workspace = true }
 edition = "2021"
 license = "GPL-3.0-only"
@@ -10,15 +10,18 @@ async-io = { workspace = true }
 async-trait = { workspace = true }
 futures = { workspace = true }
 jsonrpsee = { workspace = true, features = [ "ws-client" ] }
+schnellru = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = [ "sync" ] }
+tokio-stream = { workspace = true }
 tracing = { workspace = true }
 url = { workspace = true }
 
 # Dancekit
 dc-orchestrator-chain-interface = { workspace = true }
+dp-core = { workspace = true }
 
 # Substrate
 parity-scale-codec = { workspace = true }

client/orchestrator-chain-rpc-interface/src/lib.rs

@@ -22,7 +22,7 @@ use {
     dc_orchestrator_chain_interface::{
         OrchestratorChainError, OrchestratorChainInterface, OrchestratorChainResult, PHash, PHeader,
     },
-    futures::Stream,
+    futures::{Stream, StreamExt},
     jsonrpsee::{core::params::ArrayParams, rpc_params},
     sc_client_api::{StorageData, StorageProof},
     sc_rpc_api::state::ReadProof,
@@ -37,6 +37,7 @@ use {
 };
 
 const LOG_TARGET: &str = "orchestrator-rpc-client";
+const NOTIFICATION_CHANNEL_SIZE_LIMIT: usize = 20;
 
 /// Format url and force addition of a port
 fn url_to_string_with_port(url: Url) -> Option<String> {
@@ -138,6 +139,17 @@ impl OrchestratorChainRpcClient {
         ).await
     }
 
+    fn send_register_message(
+        &self,
+        message_builder: impl FnOnce(mpsc::Sender<dp_core::Header>) -> WsClientRequest,
+    ) -> OrchestratorChainResult<mpsc::Receiver<dp_core::Header>> {
+        let (tx, rx) = mpsc::channel(NOTIFICATION_CHANNEL_SIZE_LIMIT);
+        self.request_sender
+            .try_send(message_builder(tx))
+            .map_err(|e| OrchestratorChainError::WorkerCommunicationError(e.to_string()))?;
+        Ok(rx)
+    }
+
     /// Send a request to the RPC worker and awaits for a response. The worker is responsible
     /// for retrying requests if connection dies.
     async fn request_tracing<'a, R, OR>(
@@ -247,20 +259,26 @@ impl OrchestratorChainInterface for OrchestratorChainRpcClient {
     async fn import_notification_stream(
         &self,
     ) -> OrchestratorChainResult<Pin<Box<dyn Stream<Item = PHeader> + Send>>> {
-        todo!()
+        let rx = self.send_register_message(WsClientRequest::RegisterImportListener)?;
+        let stream = tokio_stream::wrappers::ReceiverStream::new(rx);
+        Ok(stream.boxed())
     }
 
     /// Get a stream of new best block notifications.
     async fn new_best_notification_stream(
         &self,
     ) -> OrchestratorChainResult<Pin<Box<dyn Stream<Item = PHeader> + Send>>> {
-        todo!()
+        let rx = self.send_register_message(WsClientRequest::RegisterBestHeadListener)?;
+        let stream = tokio_stream::wrappers::ReceiverStream::new(rx);
+        Ok(stream.boxed())
     }
 
     /// Get a stream of finality notifications.
     async fn finality_notification_stream(
         &self,
     ) -> OrchestratorChainResult<Pin<Box<dyn Stream<Item = PHeader> + Send>>> {
-        todo!()
+        let rx = self.send_register_message(WsClientRequest::RegisterFinalizationListener)?;
+        let stream = tokio_stream::wrappers::ReceiverStream::new(rx);
+        Ok(stream.boxed())
     }
 }