cotugno/conduit
2
Fork 0
mirror of https://gitlab.com/famedly/conduit.git synced 2024-11-17 10:48:18 -07:00
Code Issues Packages Projects Releases Wiki Activity

improvement: deduplicate watchers

Browse Source
This commit is contained in:
Timo Kösters 2021-09-13 23:19:00 +02:00
parent 06b0c9267f
commit cbee7fe111
No known key found for this signature in database
GPG Key ID: 24DA7517711A2BA4
1 changed files with 14 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
src/database/abstraction/sqlite.rs
View File

@ -4,14 +4,14 @@ use parking_lot::{Mutex, MutexGuard, RwLock};
use rusqlite::{Connection, DatabaseName::Main, OptionalExtension};
use std::{
cell::RefCell,
collections::HashMap,
collections::{hash_map, HashMap},
future::Future,
path::{Path, PathBuf},
pin::Pin,
sync::Arc,
};
use thread_local::ThreadLocal;
use tokio::sync::oneshot::Sender;
use tokio::sync::watch;
use tracing::debug;
thread_local! {
@ -126,7 +126,7 @@ impl DatabaseEngine for Engine {
pub struct SqliteTable {
engine: Arc<Engine>,
name: String,
watchers: RwLock<HashMap<Vec<u8>, Vec<Sender<()>>>>,
watchers: RwLock<HashMap<Vec<u8>, (watch::Sender<()>, watch::Receiver<()>)>>,
}
type TupleOfBytes = (Vec<u8>, Vec<u8>);
@ -215,10 +215,8 @@ impl Tree for SqliteTable {
if !triggered.is_empty() {
let mut watchers = self.watchers.write();
for prefix in triggered {
if let Some(txs) = watchers.remove(prefix) {
for tx in txs {
let _ = tx.send(());
}
if let Some(tx) = watchers.remove(prefix) {
let _ = tx.0.send(());
}
}
};
@ -367,17 +365,18 @@ impl Tree for SqliteTable {
#[tracing::instrument(skip(self, prefix))]
fn watch_prefix<'a>(&'a self, prefix: &[u8]) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>> {
let (tx, rx) = tokio::sync::oneshot::channel();
self.watchers
.write()
.entry(prefix.to_vec())
.or_default()
.push(tx);
let mut rx = match self.watchers.write().entry(prefix.to_vec()) {
hash_map::Entry::Occupied(o) => o.get().1.clone(),
hash_map::Entry::Vacant(v) => {
let (tx, rx) = tokio::sync::watch::channel(());
v.insert((tx, rx.clone()));
rx
}
};
Box::pin(async move {
// Tx is never destroyed
rx.await.unwrap();
rx.changed().await.unwrap();
})
}