Add ability to listen to node events from the frontend

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,40 @@
+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() {
+        log::debug!("node: not subscribing to events due to stopped node.");
+        return Ok(());
+    };
+
+    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 _result = 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,6 +2,7 @@
   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";
@@ -10,6 +11,10 @@
   import AuthenticationError from "@app/views/AuthenticationError.svelte";
   import Home from "@app/views/Home.svelte";
 
+  void listen("event", event => {
+    console.log(event.payload);
+  });
+
   const activeRouteStore = router.activeRouteStore;
 
   onMount(async () => {

src/lib/router.ts

@@ -5,6 +5,7 @@ import { get, writable } from "svelte/store";
 import * as mutexExecutor from "@app/lib/mutexExecutor";
 import * as utils from "@app/lib/utils";
 import { loadRoute } from "@app/lib/router/definitions";
+import { emit } from "@tauri-apps/api/event";
 
 export { type Route };
 
@@ -63,6 +64,9 @@ async function navigate(
   isLoading.set(true);
   const path = routeToPath(newRoute);
 
+  // In case there is a running subscription this won't launch a new one.
+  void emit("subscribe_events");
+
   if (action === "push") {
     window.history.pushState(newRoute, "", path);
   } else if (action === "replace") {