rpc: implemented notify handler

This enables notifying a connected stream about items as they arrive at provided subscriber. A dummy example was added to Lilith.

bin/lilith/src/main.rs

@@ -34,9 +34,11 @@ use darkfi::{
             ErrorCode::{InvalidParams, MethodNotFound},
             JsonError, JsonRequest, JsonResponse, JsonResult,
         },
-        server::{listen_and_serve, RequestHandler},
+        server::{listen_and_serve, NotifyHandler, RequestHandler},
     },
+    system::{Subscriber, SubscriberPtr},
     util::{
+        async_util::sleep,
         file::{load_file, save_file},
         path::{expand_path, get_config_path},
     },
@@ -257,6 +259,28 @@ fn save_hosts(path: &Path, spawns: FxHashMap<String, Vec<String>>) {
     }
 }
 
+// Auxillary function generating all configured JSON-RPC notification handlers for Lilith
+async fn setup_dummy_notification_handlers(mut accept_url: Url, ex: Arc<smol::Executor<'_>>) {
+    info!("Generating JSON-RPC notification handlers");
+
+    // Creating a strings notification handler listening on accept_url:18999
+    if accept_url.set_port(Some(18999)).is_err() {
+        error!("Error while setting port of notification url.");
+        return
+    }
+    let subscriber: SubscriberPtr<String> = Subscriber::new();
+    let notify_handler = NotifyHandler::new(subscriber.clone()).await;
+    // Handler starts listening for connections
+    ex.spawn(listen_and_serve(accept_url, notify_handler)).detach();
+
+    // Notifications simulation
+    let message = String::from("INSPIRACIJA");
+    loop {
+        subscriber.notify(message.clone()).await;
+        sleep(10).await;
+    }
+}
+
 async_daemonize!(realmain);
 async fn realmain(args: Args, ex: Arc<smol::Executor<'_>>) -> Result<()> {
     // We use this handler to block this function after detaching all
@@ -308,7 +332,11 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'_>>) -> Result<()> {
 
     // JSON-RPC server
     info!("Starting JSON-RPC server");
-    ex.spawn(listen_and_serve(args.rpc_listen, lilith.clone())).detach();
+    ex.spawn(listen_and_serve(args.rpc_listen.clone(), lilith.clone())).detach();
+
+    // JSON-RPC notifications
+    let _ex = ex.clone();
+    ex.spawn(setup_dummy_notification_handlers(args.rpc_listen, _ex)).detach();
 
     // Wait for SIGINT
     shutdown.recv().await?;

src/rpc/server.rs

@@ -21,6 +21,7 @@ use async_std::sync::Arc;
 use async_trait::async_trait;
 use futures::{AsyncReadExt, AsyncWriteExt};
 use log::{debug, error, info, warn};
+use serde::Serialize;
 use url::Url;
 
 use super::jsonrpc::{JsonRequest, JsonResult};
@@ -29,9 +30,22 @@ use crate::{
         TcpTransport, TorTransport, Transport, TransportListener, TransportName, TransportStream,
         UnixTransport,
     },
+    system::SubscriberPtr,
     Error, Result,
 };
 
+/// Asynchronous trait implementing an internal accept function
+/// that runs inside a loop for accepting incoming JSON-RPC requests
+/// and passing them to the handler trait.
+#[async_trait]
+pub trait AcceptTrait {
+    async fn accept(
+        self: Arc<Self>,
+        mut stream: Box<dyn TransportStream>,
+        peer_addr: Url,
+    ) -> Result<()>;
+}
+
 /// Asynchronous trait implementing a handler for incoming JSON-RPC requests.
 /// Can be used by matching on methods and branching out to functions that
 /// handle respective methods.
@@ -40,65 +54,107 @@ pub trait RequestHandler: Sync + Send {
     async fn handle_request(&self, req: JsonRequest) -> JsonResult;
 }
 
-/// Internal accept function that runs inside a loop for accepting incoming
-/// JSON-RPC requests and passing them to the [`RequestHandler`].
-async fn accept(
-    mut stream: Box<dyn TransportStream>,
-    peer_addr: Url,
-    rh: Arc<impl RequestHandler + 'static>,
-) -> Result<()> {
-    loop {
-        // FIXME: Nasty size. 8M
-        let mut buf = vec![0; 1024 * 8192];
+#[async_trait]
+impl<T: RequestHandler> AcceptTrait for T {
+    async fn accept(
+        self: Arc<Self>,
+        mut stream: Box<dyn TransportStream>,
+        peer_addr: Url,
+    ) -> Result<()> {
+        loop {
+            // FIXME: Nasty size. 8M
+            let mut buf = vec![0; 1024 * 8192];
+
+            let n = match stream.read(&mut buf).await {
+                Ok(n) if n == 0 => {
+                    debug!(target: "jsonrpc-server", "Closed connection for {}", peer_addr);
+                    break
+                }
+                Ok(n) => n,
+                Err(e) => {
+                    error!("JSON-RPC server failed reading from {} socket: {}", peer_addr, e);
+                    debug!(target: "jsonrpc-server", "Closed connection for {}", peer_addr);
+                    break
+                }
+            };
 
-        let n = match stream.read(&mut buf).await {
-            Ok(n) if n == 0 => {
-                debug!(target: "jsonrpc-server", "Closed connection for {}", peer_addr);
-                break
-            }
-            Ok(n) => n,
-            Err(e) => {
-                error!("JSON-RPC server failed reading from {} socket: {}", peer_addr, e);
+            let r: JsonRequest = match serde_json::from_slice(&buf[0..n]) {
+                Ok(r) => {
+                    debug!(target: "jsonrpc-server", "{} --> {}", peer_addr, String::from_utf8_lossy(&buf));
+                    r
+                }
+                Err(e) => {
+                    warn!("JSON-RPC server received invalid JSON from {}: {}", peer_addr, e);
+                    debug!(target: "jsonrpc-server", "Closed connection for {}", peer_addr);
+                    break
+                }
+            };
+
+            let reply = self.handle_request(r).await;
+            let j = serde_json::to_string(&reply).unwrap();
+            debug!(target: "jsonrpc-server", "{} <-- {}", peer_addr, j);
+
+            if let Err(e) = stream.write_all(j.as_bytes()).await {
+                error!("JSON-RPC server failed writing to {} socket: {}", peer_addr, e);
                 debug!(target: "jsonrpc-server", "Closed connection for {}", peer_addr);
                 break
             }
-        };
+        }
 
-        let r: JsonRequest = match serde_json::from_slice(&buf[0..n]) {
-            Ok(r) => {
-                debug!(target: "jsonrpc-server", "{} --> {}", peer_addr, String::from_utf8_lossy(&buf));
-                r
-            }
-            Err(e) => {
-                warn!("JSON-RPC server received invalid JSON from {}: {}", peer_addr, e);
+        Ok(())
+    }
+}
+
+/// Wrapper struct to notify an incoming connection about subscription items.
+pub struct NotifyHandler<T> {
+    subscriber: SubscriberPtr<T>,
+}
+
+impl<T: Clone + Serialize> NotifyHandler<T> {
+    pub async fn new(subscriber: SubscriberPtr<T>) -> Arc<Self> {
+        Arc::new(Self { subscriber })
+    }
+}
+
+#[async_trait]
+impl<T: Clone + Send + Serialize> AcceptTrait for NotifyHandler<T> {
+    async fn accept(
+        self: Arc<Self>,
+        mut stream: Box<dyn TransportStream>,
+        peer_addr: Url,
+    ) -> Result<()> {
+        let subscription = self.subscriber.clone().subscribe().await;
+
+        loop {
+            // Listen subscription for notifications
+            let notification = subscription.receive().await;
+
+            // Push notification
+            let j = serde_json::to_string(&notification).unwrap();
+            debug!(target: "jsonrpc-server", "{} <-- {}", peer_addr, j);
+
+            if let Err(e) = stream.write_all(j.as_bytes()).await {
+                debug!(target: "jsonrpc-server", "JSON-RPC server failed writing to {} socket: {}", peer_addr, e);
                 debug!(target: "jsonrpc-server", "Closed connection for {}", peer_addr);
                 break
             }
-        };
+        }
 
-        let reply = rh.handle_request(r).await;
-        let j = serde_json::to_string(&reply).unwrap();
-        debug!(target: "jsonrpc-server", "{} <-- {}", peer_addr, j);
+        subscription.unsubscribe().await;
 
-        if let Err(e) = stream.write_all(j.as_bytes()).await {
-            error!("JSON-RPC server failed writing to {} socket: {}", peer_addr, e);
-            debug!(target: "jsonrpc-server", "Closed connection for {}", peer_addr);
-            break
-        }
+        Ok(())
     }
-
-    Ok(())
 }
 
 /// Wrapper function around [`accept()`] to take the incoming connection and
 /// pass it forward.
 async fn run_accept_loop(
     listener: Box<dyn TransportListener>,
-    rh: Arc<impl RequestHandler + 'static>,
+    handler: Arc<impl AcceptTrait + 'static>,
 ) -> Result<()> {
     while let Ok((stream, peer_addr)) = listener.next().await {
         info!("JSON-RPC server accepted connection from {}", peer_addr);
-        accept(stream, peer_addr, rh.clone()).await?;
+        handler.clone().accept(stream, peer_addr).await?;
     }
 
     Ok(())
@@ -108,7 +164,7 @@ async fn run_accept_loop(
 /// [`RequestHandler`] to handle incoming requests.
 pub async fn listen_and_serve(
     accept_url: Url,
-    rh: Arc<impl RequestHandler + 'static>,
+    handler: Arc<impl AcceptTrait + 'static>,
 ) -> Result<()> {
     debug!(target: "jsonrpc-server", "Trying to bind listener on {}", accept_url);
 
@@ -129,12 +185,12 @@ pub async fn listen_and_serve(
             match $upgrade {
                 None => {
                     info!("JSON-RPC listener bound to {}", accept_url);
-                    run_accept_loop(Box::new(listener), rh).await?;
+                    run_accept_loop(Box::new(listener), handler).await?;
                 }
                 Some(u) if u == "tls" => {
                     let tls_listener = $transport.upgrade_listener(listener)?.await?;
                     info!("JSON-RPC listener bound to {}", accept_url);
-                    run_accept_loop(Box::new(tls_listener), rh).await?;
+                    run_accept_loop(Box::new(tls_listener), handler).await?;
                 }
                 Some(u) => return Err(Error::UnsupportedTransportUpgrade(u)),
             }
@@ -167,7 +223,7 @@ pub async fn listen_and_serve(
                 error!("JSON-RPC Unix socket bind to {} failed: {}", accept_url, err);
                 return Err(Error::BindFailed(accept_url.as_str().into()))
             }
-            run_accept_loop(Box::new(listener?), rh).await?;
+            run_accept_loop(Box::new(listener?), handler).await?;
         }
         _ => unimplemented!(),
     }