arkln: support the emitting event

Signed-off-by: Vincenzo Palazzo <vincenzopalazzodev@gmail.com>

Author: vincenzopalazzo

ark-lightning/Cargo.toml

@@ -18,6 +18,7 @@ thiserror = "1.0"
 tokio = { version = "1", features = ["full"] }
 tokio-tungstenite = { version = "0.24", features = ["native-tls"] }
 
+lampo-common = { git = "https://github.com/vincenzopalazzo/lampo.rs.git" }
 nosql_db = { git = "https://github.com/vincenzopalazzo/nosql-db.git" }
 nosql_sled = { git = "https://github.com/vincenzopalazzo/nosql-db.git" }
 

ark-lightning/src/boltz/boltz.rs

@@ -18,6 +18,7 @@ use super::model::PairLimits;
 use crate::arkln::Lightning;
 use crate::arkln::RcvOptions;
 use crate::arkln::SentOptions;
+use crate::boltz::boltz_ws::SwapUpdate;
 use crate::ldk::bolt11_invoice as invoice;
 use crate::ldk::offers;
 use anyhow::Ok;
@@ -27,6 +28,7 @@ use std::collections::HashMap;
 use std::future::Future;
 use std::sync::Arc;
 use tokio::sync::RwLock;
+use tokio_tungstenite::tungstenite::http::status;
 
 #[derive(Debug, Clone, Copy)]
 pub enum Network {
@@ -52,6 +54,8 @@ pub struct BoltzLightning {
     _network: Network,
     api_url: String,
     ws_client: Arc<RwLock<BoltzWebSocketClient>>,
+
+    receiver: lampo_common::event::Subscriber<super::boltz_ws::SwapUpdate>,
 }
 
 impl BoltzLightning {
@@ -62,11 +66,13 @@ impl BoltzLightning {
         let mut ws_client = BoltzWebSocketClient::new(network.clone());
         ws_client.connect().await?;
 
+        let receiver = ws_client.subscribe();
         Ok(Self {
             client,
             _network: network,
             api_url,
             ws_client: Arc::new(RwLock::new(ws_client)),
+            receiver,
         })
     }
 
@@ -84,6 +90,57 @@ impl BoltzLightning {
         Self::new(network).await
     }
 
+    pub async fn spawn(self: Arc<Self>) {
+        let this = self.clone();
+        tokio::spawn(async move {
+            let mut receiver = this.receiver.subscribe();
+            while let Some(SwapUpdate { id, status }) = receiver.recv().await {
+                match status {
+                    SwapStatus::Created => {
+                        // Check if it is inside the storage!
+                        println!("Swap {} created!", id);
+                    }
+                    SwapStatus::TransactionMempool => {
+                        // Log it
+                        println!("Swap {} transaction in mempool!", id);
+                    }
+                    SwapStatus::TransactionConfirmed => {
+                        // make a double check with what we see on chain or in the virtual mempool
+                        println!("Swap {} failed!", id);
+                    }
+                    SwapStatus::InvoiceSet => {
+                        println!("Swap {} invoice settled!", id);
+                    }
+                    SwapStatus::InvoicePending => {
+                        println!("Swap {} invoice pending!", id);
+                    }
+                    SwapStatus::InvoicePaid => {
+                        println!("Swap {} invoice paid!", id);
+                        // TODO: we should claim vthlc here!
+                    }
+                    SwapStatus::InvoiceFailedToPay => {
+                        println!("Swap {} invoice failed to pay!", id);
+                        // We should drop the swap from the storage, and probably keep track
+                        // somehow in the failure
+                    }
+                    SwapStatus::TransactionClaimed => {
+                        println!("Swap {} transaction claimed!", id);
+                    }
+                    SwapStatus::SwapExpired => {
+                        println!("Swap {} expired!", id);
+                        // We should drop the swap from the storage, and probably keep track
+                        // somehow in the failure
+                    }
+                    SwapStatus::Error { error } => {
+                        println!("Swap {} error: {}", id, error);
+                        // We should drop the swap from the storage, and probably keep track
+                        // somehow in the failure
+                    }
+                }
+            }
+        });
+    }
+
     pub async fn get_limits(&self) -> Result<HashMap<String, PairLimits>> {
         let url = format!("{}/v2/swap/submarine", self.api_url);
         let response = self.client.get(&url).send().await?;

ark-lightning/src/boltz/boltz_ws.rs

@@ -44,6 +44,7 @@ use anyhow::Result;
 use core::fmt;
 use futures_util::SinkExt;
 use futures_util::StreamExt;
+use lampo_common::event::Subscriber;
 use serde::Deserialize;
 use serde::Serialize;
 use std::sync::Arc;
@@ -265,6 +266,11 @@ pub struct BoltzWebSocketClient {
 
     sender: Arc<Mutex<Option<mpsc::UnboundedSender<WsRequest>>>>,
     receiver: Arc<Mutex<Option<mpsc::UnboundedReceiver<WsResponse>>>>,
+
+    // Event base system, the ws when receive a new update from the
+    // `receiver` channel will emit a new event with the emitter that
+    // is the location when all the subscription are listening!
+    emitter: lampo_common::event::Emitter<SwapUpdate>,
 }
 
 impl BoltzWebSocketClient {
@@ -298,9 +304,15 @@ impl BoltzWebSocketClient {
 
             sender: Arc::new(Mutex::new(None)),
             receiver: Arc::new(Mutex::new(None)),
+
+            emitter: lampo_common::event::Emitter::default(),
         }
     }
 
+    pub fn subscribe(&self) -> Subscriber<SwapUpdate> {
+        self.emitter.subscriber()
+    }
+
     /// Establishes WebSocket connection to the Boltz server
     ///
     /// # Returns
@@ -346,12 +358,24 @@ impl BoltzWebSocketClient {
         });
 
         // Spawn task to handle incoming messages
+        let emitter_clone = self.emitter.clone();
         tokio::spawn(async move {
             while let Some(msg) = read.next().await {
                 match msg {
                     Ok(Message::Text(text)) => {
-                        if let Ok(_response) = serde_json::from_str::<WsResponse>(&text) {
+                        if let Ok(response) = serde_json::from_str::<WsResponse>(&text) {
                             // FIXME: emit a new event here for the end user!
+                            match response {
+                                WsResponse::Update { args, .. } => {
+                                    for update in args {
+                                        // Emit event for each swap update
+                                        emitter_clone.emit(update.clone());
+                                    }
+                                }
+                                _ => {
+                                    println!("Received message: {:?}", response);
+                                }
+                            }
                         }
                     }
                     Ok(Message::Close(_)) => {