use std::{convert::AsRef, fmt::Debug, io::Write, sync::Arc}; use arrayvec::ArrayVec; use conduit::{err, implement, utils::result::MapExpect, Err, Result}; use futures::{future, Future, FutureExt}; use serde::Serialize; use crate::{ keyval::KeyBuf, ser, util::{is_incomplete, map_err, or_else}, Handle, }; /// Fetch a value from the database into cache, returning a reference-handle /// asynchronously. The key is serialized into an allocated buffer to perform /// the query. #[implement(super::Map)] #[inline] pub fn qry(self: &Arc, key: &K) -> impl Future>> + Send where K: Serialize + ?Sized + Debug, { let mut buf = KeyBuf::new(); self.bqry(key, &mut buf) } /// Fetch a value from the database into cache, returning a reference-handle /// asynchronously. The key is serialized into a fixed-sized buffer to perform /// the query. The maximum size is supplied as const generic parameter. #[implement(super::Map)] #[inline] pub fn aqry(self: &Arc, key: &K) -> impl Future>> + Send where K: Serialize + ?Sized + Debug, { let mut buf = ArrayVec::::new(); self.bqry(key, &mut buf) } /// Fetch a value from the database into cache, returning a reference-handle /// asynchronously. The key is serialized into a user-supplied Writer. #[implement(super::Map)] #[tracing::instrument(skip(self, buf), level = "trace")] pub fn bqry(self: &Arc, key: &K, buf: &mut B) -> impl Future>> + Send where K: Serialize + ?Sized + Debug, B: Write + AsRef<[u8]>, { let key = ser::serialize(buf, key).expect("failed to serialize query key"); self.get(key) } /// Fetch a value from the database into cache, returning a reference-handle /// asynchronously. The key is referenced directly to perform the query. #[implement(super::Map)] #[tracing::instrument(skip(self, key), fields(%self), level = "trace")] pub fn get(self: &Arc, key: &K) -> impl Future>> + Send where K: AsRef<[u8]> + Debug + ?Sized, { use crate::pool::Get; let cached = self.get_cached(key); if matches!(cached, Err(_) | Ok(Some(_))) { return future::ready(cached.map_expect("data found in cache")).boxed(); } debug_assert!(matches!(cached, Ok(None)), "expected status Incomplete"); let cmd = Get { map: self.clone(), key: key.as_ref().into(), res: None, }; self.db.pool.execute_get(cmd).boxed() } /// Fetch a value from the database into cache, returning a reference-handle. /// The key is referenced directly to perform the query. This is a thread- /// blocking call. #[implement(super::Map)] #[tracing::instrument(skip(self, key), name = "blocking", level = "trace")] pub fn get_blocking(&self, key: &K) -> Result> where K: AsRef<[u8]> + ?Sized, { self.db .db .get_pinned_cf_opt(&self.cf(), key, &self.read_options) .map_err(map_err)? .map(Handle::from) .ok_or(err!(Request(NotFound("Not found in database")))) } /// Fetch a value from the cache without I/O. #[implement(super::Map)] #[tracing::instrument(skip(self, key), name = "cache", level = "trace")] pub(crate) fn get_cached(&self, key: &K) -> Result>> where K: AsRef<[u8]> + Debug + ?Sized, { let res = self .db .db .get_pinned_cf_opt(&self.cf(), key, &self.cache_read_options); match res { // cache hit; not found Ok(None) => Err!(Request(NotFound("Not found in database"))), // cache hit; value found Ok(Some(res)) => Ok(Some(Handle::from(res))), // cache miss; unknown Err(e) if is_incomplete(&e) => Ok(None), // some other error occurred Err(e) => or_else(e), } }