bin/tau: add deg task & respective rpc calls

bin/tau/taud/src/jsonrpc.rs

@@ -30,6 +30,7 @@ use smol::lock::{Mutex, MutexGuard};
 use tinyjson::JsonValue;
 
 use darkfi::{
+    event_graph::EventGraphPtr,
     net,
     rpc::{
         jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResult, JsonSubscriber},
@@ -55,7 +56,9 @@ pub struct JsonRpcInterface {
     workspace: Mutex<String>,
     workspaces: Arc<HashMap<String, ChaChaBox>>,
     p2p: net::P2pPtr,
+    event_graph: EventGraphPtr,
     dnet_sub: JsonSubscriber,
+    deg_sub: JsonSubscriber,
     rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
 }
 
@@ -80,6 +83,11 @@ impl RequestHandler for JsonRpcInterface {
             "ping" => return self.pong(req.id, req.params).await,
             "dnet.subscribe_events" => return self.dnet_subscribe_events(req.id, req.params).await,
             "dnet.switch" => self.dnet_switch(req.params).await,
+
+            "deg.switch" => self.deg_switch(req.id, req.params).await,
+            "deg.subscribe_events" => return self.deg_subscribe_events(req.id, req.params).await,
+            "eventgraph.get_info" => return self.eg_get_info(req.id, req.params).await,
+
             // TODO: make this optional
             "p2p.get_info" => return self.p2p_get_info(req.id, req.params).await,
             _ => return JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
@@ -100,13 +108,16 @@ impl HandlerP2p for JsonRpcInterface {
 }
 
 impl JsonRpcInterface {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         dataset_path: PathBuf,
         notify_queue_sender: smol::channel::Sender<TaskInfo>,
         nickname: String,
         workspaces: Arc<HashMap<String, ChaChaBox>>,
         p2p: net::P2pPtr,
+        event_graph: EventGraphPtr,
         dnet_sub: JsonSubscriber,
+        deg_sub: JsonSubscriber,
     ) -> Self {
         let workspace = Mutex::new(workspaces.iter().last().unwrap().0.clone());
         Self {
@@ -116,8 +127,10 @@ impl JsonRpcInterface {
             workspaces,
             notify_queue_sender,
             p2p,
+            event_graph,
             rpc_connections: Mutex::new(HashSet::new()),
             dnet_sub,
+            deg_sub,
         }
     }
 
@@ -161,6 +174,60 @@ impl JsonRpcInterface {
         self.dnet_sub.clone().into()
     }
 
+    // RPCAPI:
+    // Initializes a subscription to 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) -> TaudResult<JsonValue> {
+        let params = params.get::<Vec<JsonValue>>().unwrap();
+        if params.len() != 1 || !params[0].is_bool() {
+            return Err(TaudError::InvalidData("Invalid parameters".into()))
+        }
+
+        let switch = params[0].get::<bool>().unwrap();
+
+        if *switch {
+            self.event_graph.deg_enable().await;
+        } else {
+            self.event_graph.deg_disable().await;
+        }
+
+        Ok(JsonValue::Boolean(true))
+    }
+
+    // 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 new task and returns `true` upon success.
     // --> {"jsonrpc": "2.0", "method": "add",

bin/tau/taud/src/main.rs

@@ -484,6 +484,30 @@ 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
     //
@@ -493,7 +517,9 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'static>>) -> Res
         nickname.unwrap(),
         workspaces.clone(),
         p2p.clone(),
+        event_graph.clone(),
         json_sub,
+        deg_sub,
     ));
     let rpc_task = StoppableTask::new();
     rpc_task.clone().start(