bin/genev: add deg task and related rpc calls

bin/genev/genevd/src/main.rs

@@ -22,7 +22,10 @@ use darkfi::{
     async_daemonize, cli_desc,
     event_graph::{proto::ProtocolEventGraph, EventGraph, EventGraphPtr, NULL_ID},
     net::{settings::SettingsOpt, P2p, SESSION_ALL},
-    rpc::server::{listen_and_serve, RequestHandler},
+    rpc::{
+        jsonrpc::JsonSubscriber,
+        server::{listen_and_serve, RequestHandler},
+    },
     system::{sleep, StoppableTask},
     util::path::expand_path,
     Error, Result,
@@ -167,11 +170,39 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'static>>) -> Res
         executor.clone(),
     );
 
+    info!("Starting deg subs task");
+    let deg_sub = JsonSubscriber::new("deg.subscribe_events");
+    let deg_sub_ = deg_sub.clone();
+    let event_graph_ = event_graph.clone();
+    let deg_task = StoppableTask::new();
+    deg_task.clone().start(
+        async move {
+            let deg_sub = event_graph_.deg_subscribe().await;
+            loop {
+                let event = deg_sub.receive().await;
+                debug!("Got deg event: {:?}", event);
+                deg_sub_.notify(vec![event.into()].into()).await;
+            }
+        },
+        |res| async {
+            match res {
+                Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                Err(e) => panic!("{}", e),
+            }
+        },
+        Error::DetachedTaskStopped,
+        executor.clone(),
+    );
+
     //
     // RPC interface
     //
-    let rpc_interface =
-        Arc::new(JsonRpcInterface::new("Alolymous".to_string(), event_graph.clone(), p2p.clone()));
+    let rpc_interface = Arc::new(JsonRpcInterface::new(
+        "Alolymous".to_string(),
+        event_graph.clone(),
+        p2p.clone(),
+        deg_sub,
+    ));
     let rpc_task = StoppableTask::new();
     let rpc_interface_ = rpc_interface.clone();
     rpc_task.clone().start(
@@ -193,6 +224,7 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'static>>) -> Res
 
     info!(target: "genevd", "Stopping JSON-RPC server...");
     rpc_task.stop().await;
+    deg_task.stop().await;
 
     info!(target: "genevd", "Stopping sync loop task...");
     sync_loop_task.stop().await;

bin/genev/genevd/src/rpc.rs

@@ -27,12 +27,13 @@ use darkfi::{
     event_graph::{proto::EventPut, Event, EventGraphPtr},
     net,
     rpc::{
-        jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult},
+        jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult, JsonSubscriber},
         server::RequestHandler,
     },
     system::StoppableTaskPtr,
     util::encoding::base64,
 };
+
 use darkfi_serial::{deserialize, deserialize_async_partial, serialize_async};
 use genevd::GenEvent;
 
@@ -41,6 +42,7 @@ pub struct JsonRpcInterface {
     event_graph: EventGraphPtr,
     p2p: net::P2pPtr,
     rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
+    deg_sub: JsonSubscriber,
 }
 
 #[async_trait]
@@ -52,6 +54,12 @@ impl RequestHandler for JsonRpcInterface {
 
             "ping" => self.pong(req.id, req.params).await,
             "dnet_switch" => self.dnet_switch(req.id, req.params).await,
+
+            "deg.switch" => self.deg_switch(req.id, req.params).await,
+            "deg.subscribe_events" => self.deg_subscribe_events(req.id, req.params).await,
+
+            "eventgraph.get_info" => self.eg_get_info(req.id, req.params).await,
+
             _ => return JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
         }
     }
@@ -62,8 +70,13 @@ impl RequestHandler for JsonRpcInterface {
 }
 
 impl JsonRpcInterface {
-    pub fn new(_nickname: String, event_graph: EventGraphPtr, p2p: net::P2pPtr) -> Self {
-        Self { _nickname, event_graph, p2p, rpc_connections: Mutex::new(HashSet::new()) }
+    pub fn new(
+        _nickname: String,
+        event_graph: EventGraphPtr,
+        p2p: net::P2pPtr,
+        deg_sub: JsonSubscriber,
+    ) -> Self {
+        Self { _nickname, event_graph, p2p, rpc_connections: Mutex::new(HashSet::new()), deg_sub }
     }
 
     // RPCAPI:
@@ -90,6 +103,60 @@ impl JsonRpcInterface {
         JsonResponse::new(JsonValue::Boolean(true), id).into()
     }
 
+    // RPCAPI:
+    // Initializes a subscription to p2p deg events.
+    // Once a subscription is established, apps using eventgraph will send JSON-RPC notifications of
+    // new eventgraph events to the subscriber.
+    //
+    // --> {"jsonrpc": "2.0", "method": "deg.subscribe_events", "params": [], "id": 1}
+    // <-- {"jsonrpc": "2.0", "method": "deg.subscribe_events", "params": [`event`]}
+    pub async fn deg_subscribe_events(&self, id: u16, params: JsonValue) -> JsonResult {
+        let params = params.get::<Vec<JsonValue>>().unwrap();
+        if !params.is_empty() {
+            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
+        }
+
+        self.deg_sub.clone().into()
+    }
+
+    // RPCAPI:
+    // Activate or deactivate deg in the EVENTGRAPH.
+    // By sending `true`, deg will be activated, and by sending `false` deg
+    // will be deactivated. Returns `true` on success.
+    //
+    // --> {"jsonrpc": "2.0", "method": "deg.switch", "params": [true], "id": 42}
+    // <-- {"jsonrpc": "2.0", "result": true, "id": 42}
+    async fn deg_switch(&self, id: u16, params: JsonValue) -> JsonResult {
+        let params = params.get::<Vec<JsonValue>>().unwrap();
+        if params.len() != 1 || !params[0].is_bool() {
+            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
+        }
+
+        let switch = params[0].get::<bool>().unwrap();
+
+        if *switch {
+            self.event_graph.deg_enable().await;
+        } else {
+            self.event_graph.deg_disable().await;
+        }
+
+        JsonResponse::new(JsonValue::Boolean(true), id).into()
+    }
+
+    // RPCAPI:
+    // Get EVENTGRAPH info.
+    //
+    // --> {"jsonrpc": "2.0", "method": "deg.switch", "params": [true], "id": 42}
+    // <-- {"jsonrpc": "2.0", "result": true, "id": 42}
+    async fn eg_get_info(&self, id: u16, params: JsonValue) -> JsonResult {
+        let params_ = params.get::<Vec<JsonValue>>().unwrap();
+        if !params_.is_empty() {
+            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
+        }
+
+        self.event_graph.eventgraph_info(id, params).await
+    }
+
     // RPCAPI:
     // Add a new event
     // --> {"jsonrpc": "2.0", "method": "add", "params": [], "id": 1}

src/rpc/from_impl.rs

@@ -17,7 +17,7 @@
  */
 
 use super::util::*;
-use crate::net;
+use crate::{event_graph, net};
 
 #[cfg(feature = "net")]
 impl From<net::channel::ChannelInfo> for JsonValue {
@@ -123,3 +123,28 @@ impl From<net::dnet::DnetEvent> for JsonValue {
         }
     }
 }
+
+#[cfg(feature = "net")]
+impl From<event_graph::deg::MessageInfo> for JsonValue {
+    fn from(info: event_graph::deg::MessageInfo) -> JsonValue {
+        json_map([
+            ("info", JsonArray(info.info.into_iter().map(JsonStr).collect())),
+            ("cmd", JsonStr(info.cmd)),
+            ("time", JsonStr(info.time.0.to_string())),
+        ])
+    }
+}
+
+#[cfg(feature = "net")]
+impl From<event_graph::deg::DegEvent> for JsonValue {
+    fn from(event: event_graph::deg::DegEvent) -> JsonValue {
+        match event {
+            event_graph::deg::DegEvent::SendMessage(info) => {
+                json_map([("event", json_str("send")), ("info", info.into())])
+            }
+            event_graph::deg::DegEvent::RecvMessage(info) => {
+                json_map([("event", json_str("recv")), ("info", info.into())])
+            }
+        }
+    }
+}