tyiu/notedeck
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Switch to unified timeline cache via TimelineKinds

Browse Source 
This is a fairly large rewrite which unifies our threads, timelines and
profiles. Now all timelines have a MultiSubscriber, and can be added
and removed to columns just like Threads and Profiles.

Signed-off-by: William Casarin <jb55@jb55.com>
This commit is contained in:
William Casarin
2025-01-22 15:59:21 -08:00
parent d46e526a45
commit 0cc1d8a600
39 changed files with 1395 additions and 2055 deletions
Download Patch File Download Diff File Expand all files Collapse all files

198
crates/notedeck_columns/src/multi_subscriber.rs
View File

@@ -1,107 +1,145 @@
use enostr::{Filter, RelayPool};
use nostrdb::Ndb;
use nostrdb::{Ndb, Subscription};
use tracing::{error, info};
use uuid::Uuid;
use notedeck::UnifiedSubscription;
#[derive(Debug)]
pub struct MultiSubscriber {
filters: Vec<Filter>,
pub sub: Option<UnifiedSubscription>,
subscribers: u32,
pub filters: Vec<Filter>,
pub local_subid: Option<Subscription>,
pub remote_subid: Option<String>,
local_subscribers: u32,
remote_subscribers: u32,
}
impl MultiSubscriber {
/// Create a MultiSubscriber with an initial local subscription.
pub fn with_initial_local_sub(sub: Subscription, filters: Vec<Filter>) -> Self {
let mut msub = MultiSubscriber::new(filters);
msub.local_subid = Some(sub);
msub.local_subscribers = 1;
msub
}
pub fn new(filters: Vec<Filter>) -> Self {
Self {
filters,
sub: None,
subscribers: 0,
local_subid: None,
remote_subid: None,
local_subscribers: 0,
remote_subscribers: 0,
}
}
fn real_subscribe(
ndb: &Ndb,
pool: &mut RelayPool,
filters: Vec<Filter>,
) -> Option<UnifiedSubscription> {
let subid = Uuid::new_v4().to_string();
let sub = ndb.subscribe(&filters).ok()?;
pool.subscribe(subid.clone(), filters);
Some(UnifiedSubscription {
local: sub,
remote: subid,
})
}
pub fn unsubscribe(&mut self, ndb: &mut Ndb, pool: &mut RelayPool) {
if self.subscribers == 0 {
error!("No subscribers to unsubscribe from");
return;
}
self.subscribers -= 1;
if self.subscribers == 0 {
let sub = match self.sub {
Some(ref sub) => sub,
None => {
error!("No remote subscription to unsubscribe from");
return;
}
};
let local_sub = &sub.local;
if let Err(e) = ndb.unsubscribe(*local_sub) {
error!(
"failed to unsubscribe from object: {e}, subid:{}, {} active subscriptions",
local_sub.id(),
ndb.subscription_count()
);
} else {
info!(
"Unsubscribed from object subid:{}. {} active subscriptions",
local_sub.id(),
ndb.subscription_count()
);
}
// unsub from remote
pool.unsubscribe(sub.remote.clone());
self.sub = None;
fn unsubscribe_remote(&mut self, ndb: &Ndb, pool: &mut RelayPool) {
let remote_subid = if let Some(remote_subid) = &self.remote_subid {
remote_subid
} else {
info!(
"Locally unsubscribing. {} active ndb subscriptions. {} active subscriptions for this object",
ndb.subscription_count(),
self.subscribers,
);
self.err_log(ndb, "unsubscribe_remote: nothing to unsubscribe from?");
return;
};
pool.unsubscribe(remote_subid.clone());
self.remote_subid = None;
}
/// Locally unsubscribe if we have one
fn unsubscribe_local(&mut self, ndb: &mut Ndb) {
let local_sub = if let Some(local_sub) = self.local_subid {
local_sub
} else {
self.err_log(ndb, "unsubscribe_local: nothing to unsubscribe from?");
return;
};
match ndb.unsubscribe(local_sub) {
Err(e) => {
self.err_log(ndb, &format!("Failed to unsubscribe: {e}"));
}
Ok(_) => {
self.local_subid = None;
}
}
}
pub fn unsubscribe(&mut self, ndb: &mut Ndb, pool: &mut RelayPool) -> bool {
if self.local_subscribers == 0 && self.remote_subscribers == 0 {
self.err_log(
ndb,
"Called multi_subscriber unsubscribe when both sub counts are 0",
);
return false;
}
self.local_subscribers = self.local_subscribers.saturating_sub(1);
self.remote_subscribers = self.remote_subscribers.saturating_sub(1);
if self.local_subscribers == 0 && self.remote_subscribers == 0 {
self.info_log(ndb, "Locally unsubscribing");
self.unsubscribe_local(ndb);
self.unsubscribe_remote(ndb, pool);
self.local_subscribers = 0;
self.remote_subscribers = 0;
true
} else {
false
}
}
fn info_log(&self, ndb: &Ndb, msg: &str) {
info!(
"{msg}. {}/{}/{} active ndb/local/remote subscriptions.",
ndb.subscription_count(),
self.local_subscribers,
self.remote_subscribers,
);
}
fn err_log(&self, ndb: &Ndb, msg: &str) {
error!(
"{msg}. {}/{}/{} active ndb/local/remote subscriptions.",
ndb.subscription_count(),
self.local_subscribers,
self.remote_subscribers,
);
}
pub fn subscribe(&mut self, ndb: &Ndb, pool: &mut RelayPool) {
self.subscribers += 1;
if self.subscribers == 1 {
if self.sub.is_some() {
error!("Object is first subscriber, but it already had remote subscription");
self.local_subscribers += 1;
self.remote_subscribers += 1;
if self.remote_subscribers == 1 {
if self.remote_subid.is_some() {
self.err_log(
ndb,
"Object is first subscriber, but it already had a subscription",
);
return;
} else {
let subid = Uuid::new_v4().to_string();
pool.subscribe(subid.clone(), self.filters.clone());
self.info_log(ndb, "First remote subscription");
self.remote_subid = Some(subid);
}
}
if self.local_subscribers == 1 {
if self.local_subid.is_some() {
self.err_log(ndb, "Should not have a local subscription already");
return;
}
self.sub = Self::real_subscribe(ndb, pool, self.filters.clone());
info!(
"Remotely subscribing to object. {} total active subscriptions, {} on this object",
ndb.subscription_count(),
self.subscribers,
);
match ndb.subscribe(&self.filters) {
Ok(sub) => {
self.info_log(ndb, "First local subscription");
self.local_subid = Some(sub);
}
if self.sub.is_none() {
error!("Error subscribing remotely to object");
Err(err) => {
error!("multi_subscriber: error subscribing locally: '{err}'")
}
}
} else {
info!(
"Locally subscribing. {} total active subscriptions, {} for this object",
ndb.subscription_count(),
self.subscribers,
)
}
}
}