Introduce API to subscribe to peer connection events (#625)

* Introduce API to subscribe to peer connection events

* Add entry to CHANGELOG.md

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Node API to subscribe to peer connection events [#625](https://github.com/p2panda/aquadoggo/pull/625)
 - Support private network secured by pre-shared key [#635](https://github.com/p2panda/aquadoggo/pull/635)
 
 ### Changed

aquadoggo/src/api/api.rs

@@ -1,11 +1,18 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 use anyhow::{bail, Result};
+use tokio::sync::mpsc::Receiver;
 
 use crate::api::{migrate, LockFile};
 use crate::bus::{ServiceMessage, ServiceSender};
 use crate::context::Context;
 
+#[derive(Debug, Clone)]
+pub enum NodeEvent {
+    PeerConnected,
+    PeerDisconnected,
+}
+
 /// Interface to interact with the node in a programmatic, "low-level" way.
 #[derive(Debug)]
 pub struct NodeInterface {
@@ -42,4 +49,26 @@ impl NodeInterface {
 
         Ok(did_migration_happen)
     }
+
+    pub async fn subscribe(&self) -> Receiver<NodeEvent> {
+        let mut rx = self.tx.subscribe();
+        let (events_tx, events_rx) = tokio::sync::mpsc::channel::<NodeEvent>(256);
+
+        tokio::task::spawn(async move {
+            loop {
+                match rx.recv().await {
+                    Ok(ServiceMessage::PeerConnected(_)) => {
+                        let _ = events_tx.send(NodeEvent::PeerConnected).await;
+                    }
+                    Ok(ServiceMessage::PeerDisconnected(_)) => {
+                        let _ = events_tx.send(NodeEvent::PeerDisconnected).await;
+                    }
+                    Ok(_) => continue,
+                    Err(_) => break,
+                }
+            }
+        });
+
+        events_rx
+    }
 }

aquadoggo/src/api/mod.rs

@@ -6,7 +6,7 @@ mod config_file;
 mod lock_file;
 mod migration;
 
-pub use api::NodeInterface;
+pub use api::{NodeEvent, NodeInterface};
 pub use config_file::ConfigFile;
 pub use lock_file::LockFile;
 pub use migration::migrate;

aquadoggo/src/node.rs

@@ -2,8 +2,9 @@
 
 use anyhow::Result;
 use p2panda_rs::identity::KeyPair;
+use tokio::sync::mpsc::Receiver;
 
-use crate::api::NodeInterface;
+use crate::api::{NodeEvent, NodeInterface};
 use crate::bus::ServiceMessage;
 use crate::config::Configuration;
 use crate::context::Context;
@@ -131,4 +132,10 @@ impl Node {
     pub async fn migrate(&self, lock_file: LockFile) -> Result<bool> {
         self.api.migrate(lock_file).await
     }
+
+    /// Subscribe to channel reporting on significant node events which can be interesting for
+    /// clients, for example when peers connect or disconnect.
+    pub async fn subscribe(&self) -> Receiver<NodeEvent> {
+        self.api.subscribe().await
+    }
 }