use fnv::{FnvBuildHasher, FnvHashMap};
use std::{
rc::{Rc, Weak},
sync::RwLock,
time::Duration,
};
use sycamore::reactive::Signal;
use weak_table::WeakValueHashMap;
use crate::{cache::QueryCache, AsKeys, DataSignal, Fetcher, QueryData, Status};
#[derive(Clone)]
pub struct ClientOptions {
pub cache_expiration: Duration,
pub retries: u32,
pub retry_fn: Rc<dyn Fn(u32) -> Duration>,
}
impl Default for ClientOptions {
fn default() -> Self {
Self {
cache_expiration: Duration::from_secs(5 * 60),
retries: 3,
retry_fn: Rc::new(|retries| {
Duration::from_secs((1 ^ (2 * retries)).clamp(0, 30) as u64)
}),
}
}
}
impl ClientOptions {
pub(crate) fn merge(&self, query_options: &QueryOptions) -> ClientOptions {
Self {
cache_expiration: query_options
.cache_expiration
.unwrap_or(self.cache_expiration),
retries: query_options.retries.unwrap_or(self.retries),
retry_fn: query_options
.retry_fn
.clone()
.unwrap_or_else(|| self.retry_fn.clone()),
}
}
}
#[derive(Default)]
pub struct QueryOptions {
pub cache_expiration: Option<Duration>,
pub retries: Option<u32>,
pub retry_fn: Option<Rc<dyn Fn(u32) -> Duration>>,
}
type WeakFnvMap<T> = WeakValueHashMap<Vec<u64>, Weak<T>, FnvBuildHasher>;
#[derive(Default)]
pub struct QueryClient {
pub(crate) default_options: ClientOptions,
pub(crate) cache: RwLock<QueryCache>,
pub(crate) data_signals: RwLock<WeakFnvMap<DataSignal>>,
pub(crate) status_signals: RwLock<WeakFnvMap<Signal<Status>>>,
pub(crate) fetchers: RwLock<FnvHashMap<Vec<u64>, Fetcher>>,
}
impl QueryClient {
pub fn new(default_options: ClientOptions) -> Rc<Self> {
Rc::new(Self {
default_options,
..QueryClient::default()
})
}
pub fn invalidate_queries(self: Rc<Self>, queries: Vec<Vec<u64>>) {
let queries = queries
.iter()
.map(|query| query.as_slice())
.collect::<Vec<_>>();
self.cache.write().unwrap().invalidate_keys(&queries);
log::info!(
"Invalidating queries: {queries:?}. Queries in cache: {:?}",
self.data_signals.read().unwrap().keys().collect::<Vec<_>>()
);
for query in self
.data_signals
.read()
.unwrap()
.keys()
.filter(|k| queries.iter().any(|key| k.starts_with(key)))
{
log::info!("Updating query {query:?}");
if let Some((data, status, fetcher)) = self.find_query(query, false) {
log::info!("Query present. Running fetch.");
self.clone()
.run_query(query, data, status, fetcher, &QueryOptions::default());
}
}
}
pub fn collect_garbage(&self) {
self.cache.write().unwrap().collect_garbage();
let queries = self.status_signals.read().unwrap();
self.fetchers
.write()
.unwrap()
.retain(|k, _| queries.contains_key(k));
}
pub fn query_data<K: AsKeys, T: 'static>(&self, key: K) -> Option<Rc<T>> {
let data = self.cache.read().unwrap().get(&key.as_keys())?;
Some(data.clone().downcast().unwrap())
}
pub fn set_query_data<K: AsKeys, T: 'static>(&self, key: K, value: T) {
let key = key.as_keys();
let value = Rc::new(value);
if let Some(data) = self.data_signals.read().unwrap().get(&key) {
data.set(QueryData::Ok(value.clone()))
}
self.cache
.write()
.unwrap()
.insert(key, Rc::new(value), &self.default_options);
}
}