bin/darkirc: add deg task & respective rpc calls

bin/darkirc/src/main.rs

@@ -119,6 +119,8 @@ pub struct DarkIrc {
     rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
     /// dnet JSON-RPC subscriber
     dnet_sub: JsonSubscriber,
+    /// deg JSON-RPC subscriber
+    deg_sub: JsonSubscriber,
 }
 
 impl DarkIrc {
@@ -127,8 +129,16 @@ impl DarkIrc {
         sled: sled::Db,
         event_graph: EventGraphPtr,
         dnet_sub: JsonSubscriber,
+        deg_sub: JsonSubscriber,
     ) -> Self {
-        Self { p2p, sled, event_graph, rpc_connections: Mutex::new(HashSet::new()), dnet_sub }
+        Self {
+            p2p,
+            sled,
+            event_graph,
+            rpc_connections: Mutex::new(HashSet::new()),
+            dnet_sub,
+            deg_sub,
+        }
     }
 }
 
@@ -220,9 +230,38 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
         ex.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,
+        ex.clone(),
+    );
+
     info!("Starting JSON-RPC server");
-    let darkirc =
-        Arc::new(DarkIrc::new(p2p.clone(), sled_db.clone(), event_graph.clone(), dnet_sub));
+    let darkirc = Arc::new(DarkIrc::new(
+        p2p.clone(),
+        sled_db.clone(),
+        event_graph.clone(),
+        dnet_sub,
+        deg_sub,
+    ));
     let darkirc_ = Arc::clone(&darkirc);
     let rpc_task = StoppableTask::new();
     rpc_task.clone().start(

bin/darkirc/src/rpc.rs

@@ -44,6 +44,11 @@ impl RequestHandler for DarkIrc {
             "dnet.subscribe_events" => self.dnet_subscribe_events(req.id, req.params).await,
             // TODO: Make this optional
             "p2p.get_info" => self.p2p_get_info(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,
+
             _ => JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
         }
     }
@@ -93,6 +98,60 @@ impl DarkIrc {
 
         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) -> 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
+    }
 }
 
 impl HandlerP2p for DarkIrc {