Track open subscriptions and show in stats bar

src/globals.rs

@@ -84,8 +84,12 @@ pub struct Globals {
     /// UI status messages
     pub status_queue: PRwLock<StatusQueue>,
 
+    /// How many data bytes have been read from the network, not counting overhead
     pub bytes_read: AtomicUsize,
 
+    /// How many subscriptions are open and not yet at EOSE
+    pub open_subscriptions: AtomicUsize,
+
     /// Delegation handling
     pub delegation: Delegation,
 
@@ -151,6 +155,7 @@ lazy_static! {
                 "Welcome to Gossip. Status messages will appear here. Click them to dismiss them.".to_owned()
             )),
             bytes_read: AtomicUsize::new(0),
+            open_subscriptions: AtomicUsize::new(0),
             delegation: Delegation::default(),
             media: Media::new(),
             people_search_results: PRwLock::new(Vec::new()),

src/overlord/minion/subscription.rs

@@ -1,20 +1,25 @@
+use crate::globals::GLOBALS;
 use nostr_types::{ClientMessage, Filter, SubscriptionId};
+use std::sync::atomic::Ordering;
 
-#[derive(Clone, Debug)]
+#[derive(Debug)]
 pub struct Subscription {
     id: String,
     job_id: u64,
     filters: Vec<Filter>,
     eose: bool,
+    clone: bool,
 }
 
 impl Subscription {
     pub fn new(id: &str, job_id: u64) -> Subscription {
+        GLOBALS.open_subscriptions.fetch_add(1, Ordering::SeqCst);
         Subscription {
             id: id.to_owned(),
             job_id,
             filters: vec![],
             eose: false,
+            clone: false,
         }
     }
 
@@ -37,6 +42,9 @@ impl Subscription {
     }
 
     pub fn set_eose(&mut self) {
+        if !self.clone && !self.eose {
+            GLOBALS.open_subscriptions.fetch_sub(1, Ordering::SeqCst);
+        }
         self.eose = true;
     }
 
@@ -52,3 +60,23 @@ impl Subscription {
         ClientMessage::Close(SubscriptionId(self.get_id()))
     }
 }
+
+impl Clone for Subscription {
+    fn clone(&self) -> Self {
+        Subscription {
+            id: self.id.clone(),
+            job_id: self.job_id,
+            filters: self.filters.clone(),
+            eose: self.eose,
+            clone: true,
+        }
+    }
+}
+
+impl Drop for Subscription {
+    fn drop(&mut self) {
+        if !self.clone && !self.eose {
+            GLOBALS.open_subscriptions.fetch_sub(1, Ordering::SeqCst);
+        }
+    }
+}

src/ui/mod.rs

@@ -838,11 +838,13 @@ impl eframe::App for GossipUi {
                             .unwrap_or(0);
                         let relays = GLOBALS.connected_relays.len();
                         let processed = GLOBALS.events_processed.load(Ordering::Relaxed);
+                        let subs = GLOBALS.open_subscriptions.load(Ordering::Relaxed);
                         let stats_message = format!(
-                            "EVENTS PROCESSED: {}   EVENTS: {}   RELAYS CONNECTED: {}   HTTP: {} / {}",
+                            "EVENTS: PROCESSED={} STORED={}   RELAYS CONNS={} SUBS={}    HTTP: {} / {}",
                             processed,
                             events,
                             relays,
+                            subs,
                             in_flight,
                             in_flight + queued
                         );