rpc: Support JsonSubscriber with additional JsonResponse included.

src/rpc/client.rs

@@ -123,7 +123,7 @@ impl RpcClient {
 
         // Handle the response
         match reply {
-            JsonResult::Response(rep) => {
+            JsonResult::Response(rep) | JsonResult::SubscriberWithReply(_, rep) => {
                 debug!(target: "rpc::client", "<-- {}", rep.stringify()?);
 
                 // Check if the IDs match
@@ -205,7 +205,7 @@ impl RpcClient {
                     return Err(Error::JsonRpcError((e.error.code, e.error.message)))
                 }
 
-                JsonResult::Response(r) => {
+                JsonResult::Response(r) | JsonResult::SubscriberWithReply(_, r) => {
                     debug!(target: "rpc::client", "<-- {}", r.stringify()?);
                     let e = JsonError::new(ErrorCode::InvalidReply, None, req_id);
                     return Err(Error::JsonRpcError((e.error.code, e.error.message)))

src/rpc/jsonrpc.rs

@@ -92,6 +92,7 @@ pub enum JsonResult {
     Notification(JsonNotification),
     /// Subscriber is a special object that yields a channel
     Subscriber(JsonSubscriber),
+    SubscriberWithReply(JsonSubscriber, JsonResponse),
     Request(JsonRequest),
 }
 
@@ -137,6 +138,12 @@ impl From<JsonSubscriber> for JsonResult {
     }
 }
 
+impl From<(JsonSubscriber, JsonResponse)> for JsonResult {
+    fn from(tuple: (JsonSubscriber, JsonResponse)) -> Self {
+        Self::SubscriberWithReply(tuple.0, tuple.1)
+    }
+}
+
 // ANCHOR: jsonrequest
 /// A JSON-RPC request object
 #[derive(Clone, Debug)]

src/rpc/server.rs

@@ -118,6 +118,29 @@ pub async fn accept(
                 }
             }
 
+            JsonResult::SubscriberWithReply(subscriber, reply) => {
+                // Write the response
+                debug!(target: "rpc::server", "{} <-- {}", addr, reply.stringify()?);
+                if let Err(e) = write_to_stream(&mut stream, &reply.into()).await {
+                    return Err(e)
+                }
+
+                // Start the subscriber loop
+                let subscription = subscriber.sub.subscribe().await;
+                loop {
+                    // Listen for notifications
+                    let notification = subscription.receive().await;
+
+                    // Push notification
+                    debug!(target: "rpc::server", "{} <-- {}", addr, notification.stringify()?);
+                    let notification = JsonResult::Notification(notification);
+                    if let Err(e) = write_to_stream(&mut stream, &notification).await {
+                        subscription.unsubscribe().await;
+                        return Err(e)
+                    }
+                }
+            }
+
             JsonResult::Request(_) | JsonResult::Notification(_) => {
                 unreachable!("Should never happen")
             }