8731506 Improve the event threads in the backend — radicle-desktop

deleted src-tauri/src/events.rs

@@ -1,42 +0,0 @@

 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(())
 }

modified src-tauri/src/lib.rs

@@ -1,13 +1,9 @@

 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::Emitter;
 use tauri::Manager;
 use radicle::identity::doc::PayloadId;

@@ -15,12 +11,13 @@ use radicle::identity::DocAt;

 use radicle::identity::RepoId;
 use radicle::issue::cache::Issues;
 use radicle::node::routing::Store;
 use radicle::node::Handle;
 use radicle::patch::cache::Patches;
 use radicle::storage::git::Repository;
 use radicle::storage::{ReadRepository, ReadStorage};
 use radicle::Node;
 use commands::{auth, cobs, profile, repos};
 use events::subscribe_events;
 use types::repo::SupportedPayloads;
 struct AppState {

@@ -103,22 +100,35 @@ pub fn run() {

                 }),
             }?;
             app.manage(AppState {
                 profile: profile.clone(),
             });
             let events_handler = app.handle().clone();
             let node_handler = app.handle().clone();
             let existing_events_thread: Arc<AtomicBool> = Arc::new(AtomicBool::new(false));
             let node = Node::new(profile.socket());
             let node_status = node.clone();
             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;
                 });
             app.manage(AppState { profile });
             tauri::async_runtime::spawn(async move {
                 loop {
                     let _ = node_handler.emit("node_running", node_status.is_running());
                     std::thread::sleep(std::time::Duration::from_secs(2));
                 }
             });
             tauri::async_runtime::spawn(async move {
                 loop {
                     if node.is_running() {
                         log::debug!("node: spawned node event subscription.");
                         while let Ok(events) = node.subscribe(std::time::Duration::MAX) {
                             for event in events.into_iter().flatten() {
                                 let _ = events_handler.emit("event", event);
                             }
                         }
                         log::debug!("node: event subscription loop has exited.");
                     }
                     std::thread::sleep(std::time::Duration::from_secs(2));
                 }
             });
             Ok(())

modified src/App.svelte

@@ -4,21 +4,27 @@

   import { invoke } from "@tauri-apps/api/core";
   import * as router from "@app/lib/router";
   import { nodeRunning } from "@app/lib/events";
   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";
   import Home from "@app/views/Home.svelte";
   import Issues from "@app/views/repo/Issues.svelte";
   import Patches from "@app/views/repo/Patches.svelte";
   subscribeToNodeEvents();
   import { listen } from "@tauri-apps/api/event";
   const activeRouteStore = router.activeRouteStore;
   onMount(async () => {
     try {
       await listen("event", event => {
         console.log(event.payload);
       });
       await listen<boolean>("node_running", event => {
         nodeRunning.set(event.payload);
       });
       // For development purposes don't run tauri commands when viewing from
       // a browser.
       if (window.__TAURI_INTERNALS__) {

modified src/components/Header.svelte

@@ -1,5 +1,5 @@

 <script lang="ts">
   import { nodeState } from "@app/lib/events";
   import { nodeRunning } from "@app/lib/events";
   import Border from "./Border.svelte";
   import Icon from "./Icon.svelte";

@@ -73,7 +73,7 @@

       <div class="global-flex" style:gap="0.5rem">
         <OutlineButton variant="ghost">
           {#if $nodeState === "running"}
           {#if $nodeRunning}
             <Icon name="online" />
             Online
           {:else}

modified src/lib/events.ts

@@ -1,27 +1,3 @@

 import { emit } from "@tauri-apps/api/event";
 import { listen } from "@tauri-apps/api/event";
 import { writable } from "svelte/store";
 let interval: ReturnType<typeof setInterval> | undefined = undefined;
 type NodeState = "stopped" | "running";
 export const nodeState = writable<NodeState>("stopped");
 export function subscribeToNodeEvents() {
   void listen("event", event => {
     console.log(event.payload);
   });
   void listen("node_status", event => {
     nodeState.set(event.payload as NodeState);
   });
   if (interval === undefined) {
     void emit("subscribe_events");
     interval = setInterval(() => {
       // In case there is a running subscription this won't launch a new one.
       void emit("subscribe_events");
     }, 10_000);
   }
 }
 export const nodeRunning = writable<boolean>(false);