Add ability to listen to node events from the frontend

src-tauri/Cargo.lock

@@ -3672,6 +3672,7 @@ name = "radicle-desktop"
 version = "0.0.0"
 dependencies = [
  "anyhow",
+ "log",
  "radicle",
  "radicle-surf",
  "serde",

src-tauri/Cargo.toml

@@ -20,6 +20,7 @@ tauri-build = { version = "2.0.0-rc.0", features = ["isolation"] }
 
 [dependencies]
 anyhow = { version = "1.0.86" }
+log = { version = "0.4" }
 radicle = { version = "0.13.0" }
 radicle-surf = { version = "0.22.0" }
 serde = { version = "1.0", features = ["derive"] }

src-tauri/src/error.rs

@@ -73,6 +73,10 @@ pub enum Error {
         hint: &'static str,
     },
 
+    /// Tauri error.
+    #[error(transparent)]
+    Tauri(#[from] tauri::Error),
+
     /// Storage error.
     #[error(transparent)]
     Storage(#[from] radicle::storage::Error),

src-tauri/src/events.rs

@@ -0,0 +1,42 @@
+use std::sync::{atomic::AtomicBool, Arc};
+
+use radicle::{node::Handle, Node, Profile};
+use tauri::{AppHandle, Emitter};
+
+use crate::error::Error;
+
+pub async fn subscribe_events(
+    handle: &AppHandle,
+    profile: Profile,
+    existing_events_thread: Arc<AtomicBool>,
+) -> Result<(), Error> {
+    let event_handler = handle.clone();
+    let node = Node::new(profile.socket());
+    if !node.is_running() {
+        event_handler.emit("node_status", "stopped")?;
+        log::debug!("node: not subscribing to events due to stopped node.");
+        return Ok(());
+    };
+    event_handler.emit("node_status", "running")?;
+
+    if existing_events_thread.load(std::sync::atomic::Ordering::SeqCst) {
+        log::debug!("node: not subscribing to events due to a running subscription.");
+        return Ok(());
+    } else {
+        let join_handle = tauri::async_runtime::spawn(async move {
+            log::debug!("node: spawned node event subscription.");
+            while let Ok(events) = node.subscribe(std::time::Duration::MAX) {
+                existing_events_thread.store(true, std::sync::atomic::Ordering::SeqCst);
+                for event in events.into_iter().flatten() {
+                    let _ = event_handler.emit("event", event);
+                }
+            }
+            existing_events_thread.store(false, std::sync::atomic::Ordering::SeqCst);
+            log::debug!("node: event subscription loop has exited.");
+        });
+
+        join_handle.await?;
+    }
+
+    Ok(())
+}

src-tauri/src/lib.rs

@@ -1,8 +1,13 @@
 mod commands;
 mod error;
+mod events;
 mod types;
 
+use std::sync::atomic::AtomicBool;
+use std::sync::Arc;
+
 use serde_json::json;
+use tauri::Listener;
 use tauri::Manager;
 
 use radicle::identity::doc::PayloadId;
@@ -15,6 +20,7 @@ use radicle::storage::git::Repository;
 use radicle::storage::{ReadRepository, ReadStorage};
 
 use commands::{auth, cobs, profile, repos};
+use events::subscribe_events;
 use types::repo::SupportedPayloads;
 
 struct AppState {
@@ -97,7 +103,23 @@ pub fn run() {
                 }),
             }?;
 
-            app.manage(AppState { profile });
+            app.manage(AppState {
+                profile: profile.clone(),
+            });
+
+            let existing_events_thread: Arc<AtomicBool> = Arc::new(AtomicBool::new(false));
+
+            let events_handler = app.handle().clone();
+            app.listen("subscribe_events", move |_| {
+                let profile = profile.clone();
+                let existing_events_thread = existing_events_thread.clone();
+
+                let events_handler = events_handler.to_owned();
+                tauri::async_runtime::spawn(async move {
+                    let _result =
+                        subscribe_events(&events_handler, profile, existing_events_thread).await;
+                });
+            });
 
             Ok(())
         })

src/App.svelte

@@ -2,9 +2,11 @@
   import { onMount } from "svelte";
 
   import { invoke } from "@tauri-apps/api/core";
+  import { listen } from "@tauri-apps/api/event";
 
   import * as router from "@app/lib/router";
   import { theme } from "@app/components/ThemeSwitch.svelte";
+  import { subscribeToNodeEvents } from "@app/lib/events";
   import { unreachable } from "@app/lib/utils";
 
   import AuthenticationError from "@app/views/AuthenticationError.svelte";
@@ -12,6 +14,16 @@
   import Issues from "@app/views/repo/Issues.svelte";
   import Patches from "@app/views/repo/Patches.svelte";
 
+  subscribeToNodeEvents();
+
+  void listen("event", event => {
+    console.log(event.payload);
+  });
+
+  void listen("node_status", event => {
+    console.log(`Node: ${event.payload}`);
+  });
+
   const activeRouteStore = router.activeRouteStore;
 
   onMount(async () => {

src/lib/events.ts

@@ -0,0 +1,12 @@
+import { emit } from "@tauri-apps/api/event";
+
+let interval: ReturnType<typeof setInterval> | undefined = undefined;
+
+export function subscribeToNodeEvents() {
+  if (interval === undefined) {
+    interval = setInterval(() => {
+      // In case there is a running subscription this won't launch a new one.
+      void emit("subscribe_events");
+    }, 10_000);
+  }
+}