dandri/continuwuity
1
0
Fork 0
mirror of https://forgejo.ellis.link/continuwuation/continuwuity/ synced 2026-08-28 20:08:19 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

style: Eliminate implement database

Browse Source
This commit is contained in:
timedout
2026-06-27 18:36:44 +01:00
parent 85122c62cd
commit e7b302955f
25 changed files with 1480 additions and 1520 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/database/engine/backup.rs
+70 -74
View File
@@ -1,92 +1,88 @@
use std::{ffi::OsString, path::PathBuf};
use conduwuit::{Err, Result, error, implement, info, utils::time::rfc2822_from_seconds, warn};
use conduwuit::{Err, Result, error, info, utils::time::rfc2822_from_seconds, warn};
use rocksdb::backup::{BackupEngine, BackupEngineOptions};
use super::Engine;
use crate::util::map_err;
#[implement(Engine)]
#[tracing::instrument(skip(self), level = "info")]
pub fn backup(&self) -> Result {
let mut engine = self.backup_engine()?;
let config = &self.ctx.server.config;
if config.database_backups_to_keep > 0 {
engine
.create_new_backup_flush(&self.db, true)
.map_err(map_err)?;
impl super::Engine {
#[tracing::instrument(skip(self), level = "info")]
pub fn backup(&self) -> Result {
let mut engine = self.backup_engine()?;
let config = &self.ctx.server.config;
if config.database_backups_to_keep > 0 {
engine
.create_new_backup_flush(&self.db, true)
.map_err(map_err)?;
let engine_info = engine.get_backup_info();
let info = &engine_info.last().expect("backup engine info is not empty");
info!(
"Created database backup #{} using {} bytes in {} files",
info.backup_id, info.size, info.num_files,
);
}
if config.database_backups_to_keep >= 0 {
let keep = u32::try_from(config.database_backups_to_keep)?;
if let Err(e) = engine.purge_old_backups(keep.try_into()?) {
error!("Failed to purge old backup: {e:?}");
let engine_info = engine.get_backup_info();
let info = &engine_info.last().expect("backup engine info is not empty");
info!(
"Created database backup #{} using {} bytes in {} files",
info.backup_id, info.size, info.num_files,
);
}
if config.database_backups_to_keep >= 0 {
let keep = u32::try_from(config.database_backups_to_keep)?;
if let Err(e) = engine.purge_old_backups(keep.try_into()?) {
error!("Failed to purge old backup: {e:?}");
}
}
if config.database_backups_to_keep == 0 {
warn!("Configuration item `database_backups_to_keep` is set to 0.");
}
Ok(())
}
if config.database_backups_to_keep == 0 {
warn!("Configuration item `database_backups_to_keep` is set to 0.");
pub fn backup_list(&self) -> Result<impl Iterator<Item = String> + Send> {
let info = self.backup_engine()?.get_backup_info();
if info.is_empty() {
return Err!("No backups found.");
}
let list = info.into_iter().map(|info| {
format!(
"#{} {}: {} bytes, {} files",
info.backup_id,
rfc2822_from_seconds(info.timestamp),
info.size,
info.num_files,
)
});
Ok(list)
}
Ok(())
}
pub fn backup_count(&self) -> Result<usize> {
let info = self.backup_engine()?.get_backup_info();
#[implement(Engine)]
pub fn backup_list(&self) -> Result<impl Iterator<Item = String> + Send> {
let info = self.backup_engine()?.get_backup_info();
if info.is_empty() {
return Err!("No backups found.");
Ok(info.len())
}
let list = info.into_iter().map(|info| {
format!(
"#{} {}: {} bytes, {} files",
info.backup_id,
rfc2822_from_seconds(info.timestamp),
info.size,
info.num_files,
)
});
Ok(list)
}
#[implement(Engine)]
pub fn backup_count(&self) -> Result<usize> {
let info = self.backup_engine()?.get_backup_info();
Ok(info.len())
}
#[implement(Engine)]
fn backup_engine(&self) -> Result<BackupEngine> {
let path = self.backup_path()?;
let options = BackupEngineOptions::new(path).map_err(map_err)?;
BackupEngine::open(&options, &self.ctx.env.lock()).map_err(map_err)
}
#[implement(Engine)]
fn backup_path(&self) -> Result<OsString> {
let path = self
.ctx
.server
.config
.database_backup_path
.clone()
.map(PathBuf::into_os_string)
.unwrap_or_default();
if path.is_empty() {
return Err!(Config("database_backup_path", "Configure path to enable backups"));
fn backup_engine(&self) -> Result<BackupEngine> {
let path = self.backup_path()?;
let options = BackupEngineOptions::new(path).map_err(map_err)?;
BackupEngine::open(&options, &self.ctx.env.lock()).map_err(map_err)
}
Ok(path)
fn backup_path(&self) -> Result<OsString> {
let path = self
.ctx
.server
.config
.database_backup_path
.clone()
.map(PathBuf::into_os_string)
.unwrap_or_default();
if path.is_empty() {
return Err!(Config("database_backup_path", "Configure path to enable backups"));
}
Ok(path)
}
}
src/database/engine/files.rs
+10 -10
View File
@@ -1,15 +1,15 @@
use conduwuit::{Result, implement};
use conduwuit::Result;
use rocksdb::LiveFile as SstFile;
use super::Engine;
use crate::util::map_err;
#[implement(Engine)]
pub fn file_list(&self) -> impl Iterator<Item = Result<SstFile>> + Send + use<> {
self.db
.live_files()
.map_err(map_err)
.into_iter()
.flat_map(Vec::into_iter)
.map(Ok)
impl super::Engine {
pub fn file_list(&self) -> impl Iterator<Item = Result<SstFile>> + Send + use<> {
self.db
.live_files()
.map_err(map_err)
.into_iter()
.flat_map(Vec::into_iter)
.map(Ok)
}
}
src/database/engine/memory_usage.rs
+22 -21
View File
@@ -1,30 +1,31 @@
use std::fmt::Write;
use conduwuit::{Result, implement};
use conduwuit::Result;
use rocksdb::perf::get_memory_usage_stats;
use super::Engine;
use crate::or_else;
#[implement(Engine)]
pub fn memory_usage(&self) -> Result<String> {
let mut res = String::new();
let stats = get_memory_usage_stats(Some(&[&self.db]), Some(&[&*self.ctx.row_cache.lock()]))
.or_else(or_else)?;
let mibs = |input| f64::from(u32::try_from(input / 1024).unwrap_or(0)) / 1024.0;
writeln!(
res,
"Memory buffers: {:.2} MiB\nPending write: {:.2} MiB\nTable readers: {:.2} MiB\nRow \
cache: {:.2} MiB",
mibs(stats.mem_table_total),
mibs(stats.mem_table_unflushed),
mibs(stats.mem_table_readers_total),
mibs(u64::try_from(self.ctx.row_cache.lock().get_usage())?),
)?;
impl super::Engine {
pub fn memory_usage(&self) -> Result<String> {
let mut res = String::new();
let stats =
get_memory_usage_stats(Some(&[&self.db]), Some(&[&*self.ctx.row_cache.lock()]))
.or_else(or_else)?;
let mibs = |input| f64::from(u32::try_from(input / 1024).unwrap_or(0)) / 1024.0;
writeln!(
res,
"Memory buffers: {:.2} MiB\nPending write: {:.2} MiB\nTable readers: {:.2} MiB\nRow \
cache: {:.2} MiB",
mibs(stats.mem_table_total),
mibs(stats.mem_table_unflushed),
mibs(stats.mem_table_readers_total),
mibs(u64::try_from(self.ctx.row_cache.lock().get_usage())?),
)?;
for (name, cache) in &*self.ctx.col_cache.lock() {
writeln!(res, "{name} cache: {:.2} MiB", mibs(u64::try_from(cache.get_usage())?))?;
for (name, cache) in &*self.ctx.col_cache.lock() {
writeln!(res, "{name} cache: {:.2} MiB", mibs(u64::try_from(cache.get_usage())?))?;
}
Ok(res)
}
Ok(res)
}
src/database/engine/open.rs
+93 -94
View File
@@ -8,7 +8,7 @@
use rocksdb::{ColumnFamilyDescriptor, Options};
use super::{
Db, Engine,
Db,
cf_opts::cf_options,
db_opts::db_options,
descriptor::{self, Descriptor},
@@ -16,104 +16,103 @@
};
use crate::{Context, or_else};
#[implement(Engine)]
#[tracing::instrument(skip_all, level = "info")]
pub(crate) async fn open(ctx: Arc<Context>, desc: &[Descriptor]) -> Result<Arc<Self>> {
let server = &ctx.server;
let config = &server.config;
let path = &config.database_path;
impl super::Engine {
#[tracing::instrument(skip_all, level = "info")]
pub(crate) async fn open(ctx: Arc<Context>, desc: &[Descriptor]) -> Result<Arc<Self>> {
let server = &ctx.server;
let config = &server.config;
let path = &config.database_path;
let db_opts = db_options(config, &ctx.env.lock(), &ctx.row_cache.lock())?;
let db_opts = db_options(config, &ctx.env.lock(), &ctx.row_cache.lock())?;
let cfds = Self::configure_cfds(&ctx, &db_opts, desc)?;
let num_cfds = cfds.len();
debug!("Configured {num_cfds} column descriptors...");
let cfds = Self::configure_cfds(&ctx, &db_opts, desc)?;
let num_cfds = cfds.len();
debug!("Configured {num_cfds} column descriptors...");
let load_time = std::time::Instant::now();
if config.rocksdb_repair {
repair(&db_opts, &config.database_path)?;
let load_time = std::time::Instant::now();
if config.rocksdb_repair {
repair(&db_opts, &config.database_path)?;
}
debug!("Opening database...");
let db = Db::open_cf_descriptors(&db_opts, path, cfds).or_else(or_else)?;
info!(
columns = num_cfds,
sequence = %db.latest_sequence_number(),
time = ?load_time.elapsed(),
"Opened database."
);
Ok(Arc::new(Self {
db,
pool: ctx.pool.clone(),
ctx: ctx.clone(),
checksums: config.rocksdb_checksums,
corks: AtomicU32::new(0),
}))
}
debug!("Opening database...");
let db = Db::open_cf_descriptors(&db_opts, path, cfds).or_else(or_else)?;
#[tracing::instrument(name = "configure", skip_all, level = "debug")]
fn configure_cfds(
ctx: &Arc<Context>,
db_opts: &Options,
desc: &[Descriptor],
) -> Result<Vec<ColumnFamilyDescriptor>> {
let server = &ctx.server;
let config = &server.config;
let path = &config.database_path;
let existing = Self::discover_cfs(path, db_opts);
info!(
columns = num_cfds,
sequence = %db.latest_sequence_number(),
time = ?load_time.elapsed(),
"Opened database."
);
let creating = desc.iter().filter(|desc| !existing.contains(desc.name));
Ok(Arc::new(Self {
db,
pool: ctx.pool.clone(),
ctx: ctx.clone(),
checksums: config.rocksdb_checksums,
corks: AtomicU32::new(0),
}))
}
#[implement(Engine)]
#[tracing::instrument(name = "configure", skip_all, level = "debug")]
fn configure_cfds(
ctx: &Arc<Context>,
db_opts: &Options,
desc: &[Descriptor],
) -> Result<Vec<ColumnFamilyDescriptor>> {
let server = &ctx.server;
let config = &server.config;
let path = &config.database_path;
let existing = Self::discover_cfs(path, db_opts);
let creating = desc.iter().filter(|desc| !existing.contains(desc.name));
let missing = existing
.iter()
.filter(|&name| name != "default")
.filter(|&name| !desc.iter().any(|desc| desc.name == name));
debug!(
existing = existing.len(),
described = desc.len(),
missing = missing.clone().count(),
creating = creating.clone().count(),
"Discovered database columns"
);
missing.clone().for_each(|name| {
debug!("Found unrecognized column {name:?} in existing database.");
});
creating.map(|desc| desc.name).for_each(|name| {
debug!("Creating new column {name:?} not previously found in existing database.");
});
let missing_descriptors = missing.clone().map(|_| descriptor::DROPPED);
let cfopts: Vec<_> = desc
.iter()
.copied()
.chain(missing_descriptors)
.map(|ref desc| cf_options(ctx, db_opts.clone(), desc))
.collect::<Result<_>>()?;
let cfds: Vec<_> = desc
.iter()
.map(|desc| desc.name)
.map(ToOwned::to_owned)
.chain(missing.cloned())
.zip(cfopts.into_iter())
.map(|(name, opts)| ColumnFamilyDescriptor::new(name, opts))
.collect();
Ok(cfds)
}
#[implement(Engine)]
#[tracing::instrument(name = "discover", skip_all, level = "debug")]
fn discover_cfs(path: &Path, opts: &Options) -> BTreeSet<String> {
Db::list_cf(opts, path)
.unwrap_or_default()
.into_iter()
.collect::<BTreeSet<_>>()
let missing = existing
.iter()
.filter(|&name| name != "default")
.filter(|&name| !desc.iter().any(|desc| desc.name == name));
debug!(
existing = existing.len(),
described = desc.len(),
missing = missing.clone().count(),
creating = creating.clone().count(),
"Discovered database columns"
);
missing.clone().for_each(|name| {
debug!("Found unrecognized column {name:?} in existing database.");
});
creating.map(|desc| desc.name).for_each(|name| {
debug!("Creating new column {name:?} not previously found in existing database.");
});
let missing_descriptors = missing.clone().map(|_| descriptor::DROPPED);
let cfopts: Vec<_> = desc
.iter()
.copied()
.chain(missing_descriptors)
.map(|ref desc| cf_options(ctx, db_opts.clone(), desc))
.collect::<Result<_>>()?;
let cfds: Vec<_> = desc
.iter()
.map(|desc| desc.name)
.map(ToOwned::to_owned)
.chain(missing.cloned())
.zip(cfopts.into_iter())
.map(|(name, opts)| ColumnFamilyDescriptor::new(name, opts))
.collect();
Ok(cfds)
}
#[tracing::instrument(name = "discover", skip_all, level = "debug")]
fn discover_cfs(path: &Path, opts: &Options) -> BTreeSet<String> {
Db::list_cf(opts, path)
.unwrap_or_default()
.into_iter()
.collect::<BTreeSet<_>>()
}
}
src/database/map/clear.rs
+20 -19
View File
@@ -1,30 +1,31 @@
use std::sync::Arc;
use conduwuit::{
Result, implement,
Result,
utils::stream::{ReadyExt, TryIgnore},
};
use futures::{Stream, TryStreamExt};
use crate::keyval::Key;
/// Delete all data stored in this map. !!! USE WITH CAUTION !!!
///
/// See for_clear() with additional details.
#[implement(super::Map)]
#[tracing::instrument(level = "trace")]
pub async fn clear(self: &Arc<Self>) {
self.for_clear().ignore_err().ready_for_each(|_| ()).await;
}
impl super::Map {
/// Delete all data stored in this map. !!! USE WITH CAUTION !!!
///
/// See for_clear() with additional details.
#[tracing::instrument(level = "trace")]
pub async fn clear(self: &Arc<Self>) {
// drops your entire database cutely :3
self.for_clear().ignore_err().ready_for_each(|_| ()).await;
}
/// Delete all data stored in this map. !!! USE WITH CAUTION !!!
///
/// Provides stream of keys undergoing deletion along with any errors.
///
/// Note this operation applies to a snapshot of the data when invoked.
/// Additional data written during or after this call may be missed.
#[implement(super::Map)]
#[tracing::instrument(level = "trace")]
pub fn for_clear(self: &Arc<Self>) -> impl Stream<Item = Result<Key<'_>>> + Send {
self.raw_keys().inspect_ok(|key| self.remove(key))
/// Delete all data stored in this map. !!! USE WITH CAUTION !!!
///
/// Provides stream of keys undergoing deletion along with any errors.
///
/// Note this operation applies to a snapshot of the data when invoked.
/// Additional data written during or after this call may be missed.
#[tracing::instrument(level = "trace")]
pub fn for_clear(self: &Arc<Self>) -> impl Stream<Item = Result<Key<'_>>> + Send {
self.raw_keys().inspect_ok(|key| self.remove(key))
}
}
src/database/map/compact.rs
+35 -34
View File
@@ -23,40 +23,41 @@ pub struct Options {
pub exclusive: bool,
}
#[implement(super::Map)]
#[tracing::instrument(
name = "compact",
level = "info",
skip(self),
fields(%self),
)]
pub fn compact_blocking(&self, opts: Options) -> Result {
let mut co = CompactOptions::default();
co.set_exclusive_manual_compaction(opts.exclusive);
co.set_bottommost_level_compaction(match opts.exhaustive {
| true => BottommostLevelCompaction::Force,
| false => BottommostLevelCompaction::ForceOptimized,
});
impl super::Map {
#[tracing::instrument(
name = "compact",
skip(self),
fields(%self),
)]
pub fn compact_blocking(&self, opts: Options) -> Result {
let mut co = CompactOptions::default();
co.set_exclusive_manual_compaction(opts.exclusive);
co.set_bottommost_level_compaction(match opts.exhaustive {
| true => BottommostLevelCompaction::Force,
| false => BottommostLevelCompaction::ForceOptimized,
});
match opts.level {
| (None, None) => {
co.set_change_level(true);
co.set_target_level(-1);
},
| (None, Some(level)) => {
co.set_change_level(true);
co.set_target_level(level.try_into()?);
},
| (Some(level), None) => {
co.set_change_level(false);
co.set_target_level(level.try_into()?);
},
| (Some(_), Some(_)) => return Err!("compacting between specific levels not supported"),
match opts.level {
| (None, None) => {
co.set_change_level(true);
co.set_target_level(-1);
},
| (None, Some(level)) => {
co.set_change_level(true);
co.set_target_level(level.try_into()?);
},
| (Some(level), None) => {
co.set_change_level(false);
co.set_target_level(level.try_into()?);
},
| (Some(_), Some(_)) =>
return Err!("compacting between specific levels not supported"),
}
self.db
.db
.compact_range_cf_opt(&self.cf(), opts.range.0, opts.range.1, &co);
Ok(())
}
self.db
.db
.compact_range_cf_opt(&self.cf(), opts.range.0, opts.range.1, &co);
Ok(())
}
src/database/map/contains.rs
+82 -85
View File
@@ -3,7 +3,7 @@
use conduwuit::{
Result,
arrayvec::ArrayVec,
err, implement,
err,
utils::{future::TryExtExt, result::FlatOk},
};
use futures::FutureExt;
@@ -11,93 +11,90 @@
use crate::{keyval::KeyBuf, ser};
/// Returns true if the map contains the key.
/// - key is serialized into allocated buffer
/// - harder errors may not be reported
#[inline]
#[implement(super::Map)]
pub fn contains<K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = bool> + Send + '_ + use<'_, K>
where
K: Serialize + ?Sized + Debug,
{
let mut buf = KeyBuf::new();
self.bcontains(key, &mut buf)
}
impl super::Map {
/// Returns true if the map contains the key.
/// - key is serialized into allocated buffer
/// - harder errors may not be reported
#[inline]
pub fn contains<K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = bool> + Send + '_ + use<'_, K>
where
K: Serialize + ?Sized + Debug,
{
let mut buf = KeyBuf::new();
self.bcontains(key, &mut buf)
}
/// Returns true if the map contains the key.
/// - key is serialized into stack-buffer
/// - harder errors will panic
#[inline]
#[implement(super::Map)]
pub fn acontains<const MAX: usize, K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = bool> + Send + '_ + use<'_, MAX, K>
where
K: Serialize + ?Sized + Debug,
{
let mut buf = ArrayVec::<u8, MAX>::new();
self.bcontains(key, &mut buf)
}
/// Returns true if the map contains the key.
/// - key is serialized into stack-buffer
/// - harder errors will panic
#[inline]
pub fn acontains<const MAX: usize, K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = bool> + Send + '_ + use<'_, MAX, K>
where
K: Serialize + ?Sized + Debug,
{
let mut buf = ArrayVec::<u8, MAX>::new();
self.bcontains(key, &mut buf)
}
/// Returns true if the map contains the key.
/// - key is serialized into provided buffer
/// - harder errors will panic
#[implement(super::Map)]
#[tracing::instrument(skip(self, buf), fields(%self), level = "trace")]
pub fn bcontains<K, B>(
self: &Arc<Self>,
key: &K,
buf: &mut B,
) -> impl Future<Output = bool> + Send + '_ + use<'_, K, B>
where
K: Serialize + ?Sized + Debug,
B: Write + AsRef<[u8]>,
{
let key = ser::serialize(buf, key).expect("failed to serialize query key");
self.exists(key).is_ok()
}
/// Returns true if the map contains the key.
/// - key is serialized into provided buffer
/// - harder errors will panic
#[tracing::instrument(skip(self, buf), fields(%self), level = "trace")]
pub fn bcontains<K, B>(
self: &Arc<Self>,
key: &K,
buf: &mut B,
) -> impl Future<Output = bool> + Send + '_ + use<'_, K, B>
where
K: Serialize + ?Sized + Debug,
B: Write + AsRef<[u8]>,
{
let key = ser::serialize(buf, key).expect("failed to serialize query key");
self.exists(key).is_ok()
}
/// Returns Ok if the map contains the key.
/// - key is raw
#[inline]
#[implement(super::Map)]
pub fn exists<'a, K>(
self: &'a Arc<Self>,
key: &K,
) -> impl Future<Output = Result> + Send + 'a + use<'a, K>
where
K: AsRef<[u8]> + ?Sized + Debug + 'a,
{
self.get(key).map(|res| res.map(|_| ()))
}
/// Returns Ok if the map contains the key.
/// - key is raw
#[inline]
pub fn exists<'a, K>(
self: &'a Arc<Self>,
key: &K,
) -> impl Future<Output = Result> + Send + 'a + use<'a, K>
where
K: AsRef<[u8]> + ?Sized + Debug + 'a,
{
self.get(key).map(|res| res.map(|_| ()))
}
/// Returns Ok if the map contains the key; NotFound otherwise. Harder errors
/// may not always be reported properly.
#[implement(super::Map)]
#[tracing::instrument(skip(self, key), fields(%self), level = "trace")]
pub fn exists_blocking<K>(&self, key: &K) -> Result
where
K: AsRef<[u8]> + ?Sized + Debug,
{
self.maybe_exists(key)
.then(|| self.get_blocking(key))
.flat_ok()
.map(|_| ())
.ok_or_else(|| err!(Request(NotFound("Not found in database"))))
}
/// Returns Ok if the map contains the key; NotFound otherwise. Harder
/// errors may not always be reported properly.
#[tracing::instrument(skip(self, key), fields(%self), level = "trace")]
pub fn exists_blocking<K>(&self, key: &K) -> Result
where
K: AsRef<[u8]> + ?Sized + Debug,
{
self.maybe_exists(key)
.then(|| self.get_blocking(key))
.flat_ok()
.map(|_| ())
.ok_or_else(|| err!(Request(NotFound("Not found in database"))))
}
/// Rocksdb limits this to kBlockCacheTier internally so this is not actually a
/// blocking call; in case that changes we set this as well in our read_options.
#[implement(super::Map)]
pub(crate) fn maybe_exists<K>(&self, key: &K) -> bool
where
K: AsRef<[u8]> + ?Sized,
{
self.db
.db
.key_may_exist_cf_opt(&self.cf(), key, &self.cache_read_options)
/// Rocksdb limits this to kBlockCacheTier internally so this is not
/// actually a blocking call; in case that changes we set this as well in
/// our read_options.
pub(crate) fn maybe_exists<K>(&self, key: &K) -> bool
where
K: AsRef<[u8]> + ?Sized,
{
self.db
.db
.key_may_exist_cf_opt(&self.cf(), key, &self.cache_read_options)
}
}
src/database/map/count.rs
+58 -62
View File
@@ -1,72 +1,68 @@
use std::{fmt::Debug, future::Future, sync::Arc};
use conduwuit::implement;
use futures::stream::StreamExt;
use serde::Serialize;
/// Count the total number of entries in the map.
#[implement(super::Map)]
#[inline]
pub fn count(self: &Arc<Self>) -> impl Future<Output = usize> + Send + '_ {
self.raw_keys().count()
}
impl super::Map {
/// Count the total number of entries in the map.
#[inline]
pub fn count(self: &Arc<Self>) -> impl Future<Output = usize> + Send + '_ {
self.raw_keys().count()
}
/// Count the number of entries in the map starting from a lower-bound.
///
/// - From is a structured key
#[implement(super::Map)]
#[inline]
pub fn count_from<'a, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Future<Output = usize> + Send + 'a + use<'a, P>
where
P: Serialize + ?Sized + Debug + 'a,
{
self.keys_from_raw(from).count()
}
/// Count the number of entries in the map starting from a lower-bound.
///
/// - From is a structured key
#[inline]
pub fn count_from<'a, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Future<Output = usize> + Send + 'a + use<'a, P>
where
P: Serialize + ?Sized + Debug + 'a,
{
self.keys_from_raw(from).count()
}
/// Count the number of entries in the map starting from a lower-bound.
///
/// - From is a raw
#[implement(super::Map)]
#[inline]
pub fn raw_count_from<'a, P>(
self: &'a Arc<Self>,
from: &'a P,
) -> impl Future<Output = usize> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.raw_keys_from(from).count()
}
/// Count the number of entries in the map starting from a lower-bound.
///
/// - From is a raw
#[inline]
pub fn raw_count_from<'a, P>(
self: &'a Arc<Self>,
from: &'a P,
) -> impl Future<Output = usize> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.raw_keys_from(from).count()
}
/// Count the number of entries in the map matching a prefix.
///
/// - Prefix is structured key
#[implement(super::Map)]
#[inline]
pub fn count_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Future<Output = usize> + Send + 'a + use<'a, P>
where
P: Serialize + ?Sized + Debug + 'a,
{
self.keys_prefix_raw(prefix).count()
}
/// Count the number of entries in the map matching a prefix.
///
/// - Prefix is structured key
#[inline]
pub fn count_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Future<Output = usize> + Send + 'a + use<'a, P>
where
P: Serialize + ?Sized + Debug + 'a,
{
self.keys_prefix_raw(prefix).count()
}
/// Count the number of entries in the map matching a prefix.
///
/// - Prefix is raw
#[implement(super::Map)]
#[inline]
pub fn raw_count_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Future<Output = usize> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.raw_keys_prefix(prefix).count()
/// Count the number of entries in the map matching a prefix.
///
/// - Prefix is raw
#[inline]
pub fn raw_count_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Future<Output = usize> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.raw_keys_prefix(prefix).count()
}
}
src/database/map/get.rs
+62 -64
View File
@@ -1,6 +1,6 @@
use std::{convert::AsRef, fmt::Debug, sync::Arc};
use conduwuit::{Err, Result, err, implement, utils::result::MapExpect};
use conduwuit::{Err, Result, err, utils::result::MapExpect};
use futures::{Future, FutureExt, TryFutureExt, future::ready};
use rocksdb::{DBPinnableSlice, ReadOptions};
use tokio::task;
@@ -10,74 +10,72 @@
util::{is_incomplete, map_err, or_else},
};
/// 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<K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = Result<Handle<'_>>> + Send + use<'_, K>
where
K: AsRef<[u8]> + Debug + ?Sized,
{
use crate::pool::Get;
impl super::Map {
/// Fetch a value from the database into cache, returning a reference-handle
/// asynchronously. The key is referenced directly to perform the query.
#[tracing::instrument(skip(self, key), fields(%self), level = "trace")]
pub fn get<K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = Result<Handle<'_>>> + Send + use<'_, K>
where
K: AsRef<[u8]> + Debug + ?Sized,
{
use crate::pool::Get;
let cached = self.get_cached(key);
if matches!(cached, Err(_) | Ok(Some(_))) {
return task::consume_budget()
.map(move |()| cached.map_expect("data found in cache"))
.boxed();
let cached = self.get_cached(key);
if matches!(cached, Err(_) | Ok(Some(_))) {
return task::consume_budget()
.map(move |()| 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()].into(),
res: None,
};
self.db
.pool
.execute_get(cmd)
.and_then(|mut res| ready(res.remove(0)))
.boxed()
}
debug_assert!(matches!(cached, Ok(None)), "expected status Incomplete");
let cmd = Get {
map: self.clone(),
key: [key.as_ref().into()].into(),
res: None,
};
/// Fetch a value from the cache without I/O.
#[tracing::instrument(skip(self, key), name = "cache", level = "trace")]
pub(crate) fn get_cached<K>(&self, key: &K) -> Result<Option<Handle<'_>>>
where
K: AsRef<[u8]> + Debug + ?Sized,
{
let res = self.get_blocking_opts(key, &self.cache_read_options);
cached_handle_from(res)
}
self.db
.pool
.execute_get(cmd)
.and_then(|mut res| ready(res.remove(0)))
.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.
#[tracing::instrument(skip(self, key), name = "blocking", level = "trace")]
pub fn get_blocking<K>(&self, key: &K) -> Result<Handle<'_>>
where
K: AsRef<[u8]> + ?Sized,
{
let res = self.get_blocking_opts(key, &self.read_options);
handle_from(res)
}
/// 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<K>(&self, key: &K) -> Result<Option<Handle<'_>>>
where
K: AsRef<[u8]> + Debug + ?Sized,
{
let res = self.get_blocking_opts(key, &self.cache_read_options);
cached_handle_from(res)
}
/// 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<K>(&self, key: &K) -> Result<Handle<'_>>
where
K: AsRef<[u8]> + ?Sized,
{
let res = self.get_blocking_opts(key, &self.read_options);
handle_from(res)
}
#[implement(super::Map)]
fn get_blocking_opts<K>(
&self,
key: &K,
read_options: &ReadOptions,
) -> Result<Option<DBPinnableSlice<'_>>, rocksdb::Error>
where
K: AsRef<[u8]> + ?Sized,
{
self.db.db.get_pinned_cf_opt(&self.cf(), key, read_options)
fn get_blocking_opts<K>(
&self,
key: &K,
read_options: &ReadOptions,
) -> Result<Option<DBPinnableSlice<'_>>, rocksdb::Error>
where
K: AsRef<[u8]> + ?Sized,
{
self.db.db.get_pinned_cf_opt(&self.cf(), key, read_options)
}
}
#[inline]
src/database/map/get_batch.rs
+51 -52
View File
@@ -1,7 +1,7 @@
use std::{convert::AsRef, sync::Arc};
use conduwuit::{
Result, implement,
Result,
utils::{
IterStream,
stream::{WidebandExt, automatic_amplification, automatic_width},
@@ -34,60 +34,59 @@ fn get(self, map: &'a Arc<super::Map>) -> impl Stream<Item = Result<Handle<'a>>>
}
}
#[implement(super::Map)]
#[tracing::instrument(skip(self, keys), level = "trace")]
pub(crate) fn get_batch<'a, S, K>(
self: &'a Arc<Self>,
keys: S,
) -> impl Stream<Item = Result<Handle<'a>>> + Send + 'a
where
S: Stream<Item = K> + Send + 'a,
K: AsRef<[u8]> + Send + Sync + 'a,
{
use crate::pool::Get;
impl super::Map {
#[tracing::instrument(skip(self, keys), level = "trace")]
pub(crate) fn get_batch<'a, S, K>(
self: &'a Arc<Self>,
keys: S,
) -> impl Stream<Item = Result<Handle<'a>>> + Send + 'a
where
S: Stream<Item = K> + Send + 'a,
K: AsRef<[u8]> + Send + Sync + 'a,
{
use crate::pool::Get;
keys.ready_chunks(automatic_amplification())
.widen_then(automatic_width(), |chunk| {
self.db.pool.execute_get(Get {
map: self.clone(),
key: chunk.iter().map(AsRef::as_ref).map(Into::into).collect(),
res: None,
keys.ready_chunks(automatic_amplification())
.widen_then(automatic_width(), |chunk| {
self.db.pool.execute_get(Get {
map: self.clone(),
key: chunk.iter().map(AsRef::as_ref).map(Into::into).collect(),
res: None,
})
})
})
.map_ok(|results| results.into_iter().stream())
.try_flatten()
}
.map_ok(|results| results.into_iter().stream())
.try_flatten()
}
#[implement(super::Map)]
#[tracing::instrument(name = "batch_blocking", level = "trace", skip_all)]
pub(crate) fn get_batch_blocking<'a, I, K>(
&self,
keys: I,
) -> impl Iterator<Item = Result<Handle<'_>>> + Send + use<'_, I, K>
where
I: Iterator<Item = &'a K> + ExactSizeIterator + Send,
K: AsRef<[u8]> + Send + ?Sized + Sync + 'a,
{
self.get_batch_blocking_opts(keys, &self.read_options)
.map(handle_from)
}
#[tracing::instrument(name = "batch_blocking", level = "trace", skip_all)]
pub(crate) fn get_batch_blocking<'a, I, K>(
&self,
keys: I,
) -> impl Iterator<Item = Result<Handle<'_>>> + Send + use<'_, I, K>
where
I: Iterator<Item = &'a K> + ExactSizeIterator + Send,
K: AsRef<[u8]> + Send + ?Sized + Sync + 'a,
{
self.get_batch_blocking_opts(keys, &self.read_options)
.map(handle_from)
}
#[implement(super::Map)]
fn get_batch_blocking_opts<'a, I, K>(
&self,
keys: I,
read_options: &ReadOptions,
) -> impl Iterator<Item = Result<Option<DBPinnableSlice<'_>>, rocksdb::Error>> + Send + use<'_, I, K>
where
I: Iterator<Item = &'a K> + ExactSizeIterator + Send,
K: AsRef<[u8]> + Send + ?Sized + Sync + 'a,
{
// Optimization can be `true` if key vector is pre-sorted **by the column
// comparator**.
const SORTED: bool = false;
fn get_batch_blocking_opts<'a, I, K>(
&self,
keys: I,
read_options: &ReadOptions,
) -> impl Iterator<Item = Result<Option<DBPinnableSlice<'_>>, rocksdb::Error>> + Send + use<'_, I, K>
where
I: Iterator<Item = &'a K> + ExactSizeIterator + Send,
K: AsRef<[u8]> + Send + ?Sized + Sync + 'a,
{
// Optimization can be `true` if key vector is pre-sorted **by the column
// comparator**.
const SORTED: bool = false;
self.db
.db
.batched_multi_get_cf_opt(&self.cf(), keys, SORTED, read_options)
.into_iter()
self.db
.db
.batched_multi_get_cf_opt(&self.cf(), keys, SORTED, read_options)
.into_iter()
}
}
src/database/map/insert.rs
+203 -214
View File
@@ -5,7 +5,7 @@
use std::{convert::AsRef, fmt::Debug, io::Write};
use conduwuit::{arrayvec::ArrayVec, implement};
use conduwuit::arrayvec::ArrayVec;
use rocksdb::WriteBatchWithTransaction;
use serde::Serialize;
@@ -15,223 +15,212 @@
util::or_else,
};
/// Insert Key/Value
///
/// - Key is serialized
/// - Val is serialized
#[implement(super::Map)]
#[inline]
pub fn put<K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: Serialize,
{
let mut key_buf = KeyBuf::new();
let mut val_buf = ValBuf::new();
self.bput(key, val, (&mut key_buf, &mut val_buf));
}
/// Insert Key/Value
///
/// - Key is serialized
/// - Val is raw
#[implement(super::Map)]
#[inline]
pub fn put_raw<K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: AsRef<[u8]>,
{
let mut key_buf = KeyBuf::new();
self.bput_raw(key, val, &mut key_buf);
}
/// Insert Key/Value
///
/// - Key is raw
/// - Val is serialized
#[implement(super::Map)]
#[inline]
pub fn raw_put<K, V>(&self, key: K, val: V)
where
K: AsRef<[u8]>,
V: Serialize,
{
let mut val_buf = ValBuf::new();
self.raw_bput(key, val, &mut val_buf);
}
/// Insert Key/Value
///
/// - Key is serialized
/// - Val is serialized to stack-buffer
#[implement(super::Map)]
#[inline]
pub fn put_aput<const VMAX: usize, K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: Serialize,
{
let mut key_buf = KeyBuf::new();
let mut val_buf = ArrayVec::<u8, VMAX>::new();
self.bput(key, val, (&mut key_buf, &mut val_buf));
}
/// Insert Key/Value
///
/// - Key is serialized to stack-buffer
/// - Val is serialized
#[implement(super::Map)]
#[inline]
pub fn aput_put<const KMAX: usize, K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: Serialize,
{
let mut key_buf = ArrayVec::<u8, KMAX>::new();
let mut val_buf = ValBuf::new();
self.bput(key, val, (&mut key_buf, &mut val_buf));
}
/// Insert Key/Value
///
/// - Key is serialized to stack-buffer
/// - Val is serialized to stack-buffer
#[implement(super::Map)]
#[inline]
pub fn aput<const KMAX: usize, const VMAX: usize, K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: Serialize,
{
let mut key_buf = ArrayVec::<u8, KMAX>::new();
let mut val_buf = ArrayVec::<u8, VMAX>::new();
self.bput(key, val, (&mut key_buf, &mut val_buf));
}
/// Insert Key/Value
///
/// - Key is serialized to stack-buffer
/// - Val is raw
#[implement(super::Map)]
#[inline]
pub fn aput_raw<const KMAX: usize, K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: AsRef<[u8]>,
{
let mut key_buf = ArrayVec::<u8, KMAX>::new();
self.bput_raw(key, val, &mut key_buf);
}
/// Insert Key/Value
///
/// - Key is raw
/// - Val is serialized to stack-buffer
#[implement(super::Map)]
#[inline]
pub fn raw_aput<const VMAX: usize, K, V>(&self, key: K, val: V)
where
K: AsRef<[u8]>,
V: Serialize,
{
let mut val_buf = ArrayVec::<u8, VMAX>::new();
self.raw_bput(key, val, &mut val_buf);
}
/// Insert Key/Value
///
/// - Key is serialized to supplied buffer
/// - Val is serialized to supplied buffer
#[implement(super::Map)]
pub fn bput<K, V, Bk, Bv>(&self, key: K, val: V, mut buf: (Bk, Bv))
where
K: Serialize + Debug,
V: Serialize,
Bk: Write + AsRef<[u8]>,
Bv: Write + AsRef<[u8]>,
{
let val = ser::serialize(&mut buf.1, val).expect("failed to serialize insertion val");
self.bput_raw(key, val, &mut buf.0);
}
/// Insert Key/Value
///
/// - Key is serialized to supplied buffer
/// - Val is raw
#[implement(super::Map)]
#[tracing::instrument(skip(self, val, buf), level = "trace")]
pub fn bput_raw<K, V, Bk>(&self, key: K, val: V, mut buf: Bk)
where
K: Serialize + Debug,
V: AsRef<[u8]>,
Bk: Write + AsRef<[u8]>,
{
let key = ser::serialize(&mut buf, key).expect("failed to serialize insertion key");
self.insert(&key, val);
}
/// Insert Key/Value
///
/// - Key is raw
/// - Val is serialized to supplied buffer
#[implement(super::Map)]
pub fn raw_bput<K, V, Bv>(&self, key: K, val: V, mut buf: Bv)
where
K: AsRef<[u8]>,
V: Serialize,
Bv: Write + AsRef<[u8]>,
{
let val = ser::serialize(&mut buf, val).expect("failed to serialize insertion val");
self.insert(&key, val);
}
/// Insert Key/Value
///
/// - Key is raw
/// - Val is raw
#[implement(super::Map)]
#[tracing::instrument(skip_all, fields(%self), level = "trace")]
pub fn insert<K, V>(&self, key: &K, val: V)
where
K: AsRef<[u8]> + ?Sized,
V: AsRef<[u8]>,
{
let write_options = &self.write_options;
self.db
.db
.put_cf_opt(&self.cf(), key, val, write_options)
.or_else(or_else)
.expect("database insert error");
if !self.db.corked() {
self.db.flush().expect("database flush error");
impl super::Map {
/// Insert Key/Value
///
/// - Key is serialized
/// - Val is serialized
#[inline]
pub fn put<K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: Serialize,
{
let mut key_buf = KeyBuf::new();
let mut val_buf = ValBuf::new();
self.bput(key, val, (&mut key_buf, &mut val_buf));
}
self.watchers.wake(key.as_ref());
}
#[implement(super::Map)]
#[tracing::instrument(skip(self, iter), fields(%self), level = "trace")]
pub fn insert_batch<'a, I, K, V>(&'a self, iter: I)
where
I: Iterator<Item = (K, V)> + Send + Debug,
K: AsRef<[u8]> + Sized + Debug + 'a,
V: AsRef<[u8]> + Sized + 'a,
{
let mut batch = WriteBatchWithTransaction::<false>::default();
for (key, val) in iter {
batch.put_cf(&self.cf(), key.as_ref(), val.as_ref());
/// Insert Key/Value
///
/// - Key is serialized
/// - Val is raw
#[inline]
pub fn put_raw<K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: AsRef<[u8]>,
{
let mut key_buf = KeyBuf::new();
self.bput_raw(key, val, &mut key_buf);
}
let write_options = &self.write_options;
self.db
.db
.write_opt(&batch, write_options)
.or_else(or_else)
.expect("database insert batch error");
/// Insert Key/Value
///
/// - Key is raw
/// - Val is serialized
#[inline]
pub fn raw_put<K, V>(&self, key: K, val: V)
where
K: AsRef<[u8]>,
V: Serialize,
{
let mut val_buf = ValBuf::new();
self.raw_bput(key, val, &mut val_buf);
}
if !self.db.corked() {
self.db.flush().expect("database flush error");
/// Insert Key/Value
///
/// - Key is serialized
/// - Val is serialized to stack-buffer
#[inline]
pub fn put_aput<const VMAX: usize, K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: Serialize,
{
let mut key_buf = KeyBuf::new();
let mut val_buf = ArrayVec::<u8, VMAX>::new();
self.bput(key, val, (&mut key_buf, &mut val_buf));
}
/// Insert Key/Value
///
/// - Key is serialized to stack-buffer
/// - Val is serialized
#[inline]
pub fn aput_put<const KMAX: usize, K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: Serialize,
{
let mut key_buf = ArrayVec::<u8, KMAX>::new();
let mut val_buf = ValBuf::new();
self.bput(key, val, (&mut key_buf, &mut val_buf));
}
/// Insert Key/Value
///
/// - Key is serialized to stack-buffer
/// - Val is serialized to stack-buffer
#[inline]
pub fn aput<const KMAX: usize, const VMAX: usize, K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: Serialize,
{
let mut key_buf = ArrayVec::<u8, KMAX>::new();
let mut val_buf = ArrayVec::<u8, VMAX>::new();
self.bput(key, val, (&mut key_buf, &mut val_buf));
}
/// Insert Key/Value
///
/// - Key is serialized to stack-buffer
/// - Val is raw
#[inline]
pub fn aput_raw<const KMAX: usize, K, V>(&self, key: K, val: V)
where
K: Serialize + Debug,
V: AsRef<[u8]>,
{
let mut key_buf = ArrayVec::<u8, KMAX>::new();
self.bput_raw(key, val, &mut key_buf);
}
/// Insert Key/Value
///
/// - Key is raw
/// - Val is serialized to stack-buffer
#[inline]
pub fn raw_aput<const VMAX: usize, K, V>(&self, key: K, val: V)
where
K: AsRef<[u8]>,
V: Serialize,
{
let mut val_buf = ArrayVec::<u8, VMAX>::new();
self.raw_bput(key, val, &mut val_buf);
}
/// Insert Key/Value
///
/// - Key is serialized to supplied buffer
/// - Val is serialized to supplied buffer
pub fn bput<K, V, Bk, Bv>(&self, key: K, val: V, mut buf: (Bk, Bv))
where
K: Serialize + Debug,
V: Serialize,
Bk: Write + AsRef<[u8]>,
Bv: Write + AsRef<[u8]>,
{
let val = ser::serialize(&mut buf.1, val).expect("failed to serialize insertion val");
self.bput_raw(key, val, &mut buf.0);
}
/// Insert Key/Value
///
/// - Key is serialized to supplied buffer
/// - Val is raw
#[tracing::instrument(skip(self, val, buf), level = "trace")]
pub fn bput_raw<K, V, Bk>(&self, key: K, val: V, mut buf: Bk)
where
K: Serialize + Debug,
V: AsRef<[u8]>,
Bk: Write + AsRef<[u8]>,
{
let key = ser::serialize(&mut buf, key).expect("failed to serialize insertion key");
self.insert(&key, val);
}
/// Insert Key/Value
///
/// - Key is raw
/// - Val is serialized to supplied buffer
pub fn raw_bput<K, V, Bv>(&self, key: K, val: V, mut buf: Bv)
where
K: AsRef<[u8]>,
V: Serialize,
Bv: Write + AsRef<[u8]>,
{
let val = ser::serialize(&mut buf, val).expect("failed to serialize insertion val");
self.insert(&key, val);
}
/// Insert Key/Value
///
/// - Key is raw
/// - Val is raw
#[tracing::instrument(skip_all, fields(%self), level = "trace")]
pub fn insert<K, V>(&self, key: &K, val: V)
where
K: AsRef<[u8]> + ?Sized,
V: AsRef<[u8]>,
{
let write_options = &self.write_options;
self.db
.db
.put_cf_opt(&self.cf(), key, val, write_options)
.or_else(or_else)
.expect("database insert error");
if !self.db.corked() {
self.db.flush().expect("database flush error");
}
self.watchers.wake(key.as_ref());
}
#[tracing::instrument(skip(self, iter), fields(%self), level = "trace")]
pub fn insert_batch<'a, I, K, V>(&'a self, iter: I)
where
I: Iterator<Item = (K, V)> + Send + Debug,
K: AsRef<[u8]> + Sized + Debug + 'a,
V: AsRef<[u8]> + Sized + 'a,
{
let mut batch = WriteBatchWithTransaction::<false>::default();
for (key, val) in iter {
batch.put_cf(&self.cf(), key.as_ref(), val.as_ref());
}
let write_options = &self.write_options;
self.db
.db
.write_opt(&batch, write_options)
.or_else(or_else)
.expect("database insert batch error");
if !self.db.corked() {
self.db.flush().expect("database flush error");
}
}
}
src/database/map/keys_from.rs
+64 -66
View File
@@ -11,73 +11,71 @@
stream,
};
#[implement(super::Map)]
pub fn keys_from<'a, K, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
{
self.keys_from_raw(from).map(result_deserialize_key::<K>)
}
#[implement(super::Map)]
#[tracing::instrument(skip(self), level = "trace")]
pub fn keys_from_raw<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(from).expect("failed to serialize query key");
self.raw_keys_from(&key)
}
#[implement(super::Map)]
pub fn keys_raw_from<'a, K, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: AsRef<[u8]> + ?Sized + Debug + Sync,
K: Deserialize<'a> + Send,
{
self.raw_keys_from(from).map(result_deserialize_key::<K>)
}
#[implement(super::Map)]
#[tracing::instrument(skip(self, from), fields(%self), level = "trace")]
pub fn raw_keys_from<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: AsRef<[u8]> + ?Sized + Debug,
{
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self, from) {
return stream::Keys::<'_>::from(state.init_fwd(from.as_ref().into())).boxed();
impl super::Map {
pub fn keys_from<'a, K, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
{
self.keys_from_raw(from).map(result_deserialize_key::<K>)
}
let seek = Seek {
map: self.clone(),
dir: Direction::Forward,
key: Some(from.as_ref().into()),
state: crate::pool::into_send_seek(state),
res: None,
};
#[tracing::instrument(skip(self), level = "trace")]
pub fn keys_from_raw<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(from).expect("failed to serialize query key");
self.raw_keys_from(&key)
}
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::Keys<'_>>()
.into_stream()
.try_flatten()
.boxed()
pub fn keys_raw_from<'a, K, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: AsRef<[u8]> + ?Sized + Debug + Sync,
K: Deserialize<'a> + Send,
{
self.raw_keys_from(from).map(result_deserialize_key::<K>)
}
#[tracing::instrument(skip(self, from), fields(%self), level = "trace")]
pub fn raw_keys_from<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: AsRef<[u8]> + ?Sized + Debug,
{
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self, from) {
return stream::Keys::<'_>::from(state.init_fwd(from.as_ref().into())).boxed();
}
let seek = Seek {
map: self.clone(),
dir: Direction::Forward,
key: Some(from.as_ref().into()),
state: crate::pool::into_send_seek(state),
res: None,
};
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::Keys<'_>>()
.into_stream()
.try_flatten()
.boxed()
}
}
src/database/map/keys_prefix.rs
+46 -48
View File
@@ -1,59 +1,57 @@
use std::{convert::AsRef, fmt::Debug, sync::Arc};
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{Stream, StreamExt, TryStreamExt, future};
use serde::{Deserialize, Serialize};
use crate::keyval::{Key, result_deserialize_key, serialize_key};
#[implement(super::Map)]
pub fn keys_prefix<'a, K, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
{
self.keys_prefix_raw(prefix)
.map(result_deserialize_key::<K>)
}
impl super::Map {
pub fn keys_prefix<'a, K, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
{
self.keys_prefix_raw(prefix)
.map(result_deserialize_key::<K>)
}
#[implement(super::Map)]
#[tracing::instrument(skip(self), level = "trace")]
pub fn keys_prefix_raw<P>(
self: &Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(prefix).expect("failed to serialize query key");
self.raw_keys_from(&key)
.try_take_while(move |k: &Key<'_>| future::ok(k.starts_with(&key)))
}
#[tracing::instrument(skip(self), level = "trace")]
pub fn keys_prefix_raw<P>(
self: &Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(prefix).expect("failed to serialize query key");
self.raw_keys_from(&key)
.try_take_while(move |k: &Key<'_>| future::ok(k.starts_with(&key)))
}
#[implement(super::Map)]
pub fn keys_raw_prefix<'a, K, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
K: Deserialize<'a> + Send + 'a,
{
self.raw_keys_prefix(prefix)
.map(result_deserialize_key::<K>)
}
pub fn keys_raw_prefix<'a, K, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
K: Deserialize<'a> + Send + 'a,
{
self.raw_keys_prefix(prefix)
.map(result_deserialize_key::<K>)
}
#[implement(super::Map)]
pub fn raw_keys_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<Key<'a>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.raw_keys_from(prefix)
.try_take_while(|k: &Key<'_>| future::ok(k.starts_with(prefix.as_ref())))
pub fn raw_keys_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<Key<'a>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.raw_keys_from(prefix)
.try_take_while(|k: &Key<'_>| future::ok(k.starts_with(prefix.as_ref())))
}
}
src/database/map/qry.rs
+46 -46
View File
@@ -1,56 +1,56 @@
use std::{convert::AsRef, fmt::Debug, io::Write, sync::Arc};
use conduwuit::{Result, arrayvec::ArrayVec, implement};
use conduwuit::{Result, arrayvec::ArrayVec};
use futures::Future;
use serde::Serialize;
use crate::{Handle, keyval::KeyBuf, ser};
/// 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<K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = Result<Handle<'_>>> + Send + use<'_, K>
where
K: Serialize + ?Sized + Debug,
{
let mut buf = KeyBuf::new();
self.bqry(key, &mut buf)
}
impl super::Map {
/// 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.
#[inline]
pub fn qry<K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = Result<Handle<'_>>> + Send + use<'_, K>
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<const MAX: usize, K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = Result<Handle<'_>>> + Send + use<'_, MAX, K>
where
K: Serialize + ?Sized + Debug,
{
let mut buf = ArrayVec::<u8, MAX>::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.
#[inline]
pub fn aqry<const MAX: usize, K>(
self: &Arc<Self>,
key: &K,
) -> impl Future<Output = Result<Handle<'_>>> + Send + use<'_, MAX, K>
where
K: Serialize + ?Sized + Debug,
{
let mut buf = ArrayVec::<u8, MAX>::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<K, B>(
self: &Arc<Self>,
key: &K,
buf: &mut B,
) -> impl Future<Output = Result<Handle<'_>>> + Send + use<'_, K, B>
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 serialized into a user-supplied Writer.
#[tracing::instrument(skip(self, buf), level = "trace")]
pub fn bqry<K, B>(
self: &Arc<Self>,
key: &K,
buf: &mut B,
) -> impl Future<Output = Result<Handle<'_>>> + Send + use<'_, K, B>
where
K: Serialize + ?Sized + Debug,
B: Write + AsRef<[u8]>,
{
let key = ser::serialize(buf, key).expect("failed to serialize query key");
self.get(key)
}
}
src/database/map/qry_batch.rs
+25 -24
View File
@@ -32,30 +32,31 @@ fn qry(self, map: &'a Arc<super::Map>) -> impl Stream<Item = Result<Handle<'a>>>
}
}
#[implement(super::Map)]
#[tracing::instrument(skip(self, keys), level = "trace")]
pub(crate) fn qry_batch<'a, S, K>(
self: &'a Arc<Self>,
keys: S,
) -> impl Stream<Item = Result<Handle<'a>>> + Send + 'a
where
S: Stream<Item = K> + Send + 'a,
K: Serialize + Debug + 'a,
{
use crate::pool::Get;
impl super::Map {
#[tracing::instrument(skip(self, keys), level = "trace")]
pub(crate) fn qry_batch<'a, S, K>(
self: &'a Arc<Self>,
keys: S,
) -> impl Stream<Item = Result<Handle<'a>>> + Send + 'a
where
S: Stream<Item = K> + Send + 'a,
K: Serialize + Debug + 'a,
{
use crate::pool::Get;
keys.ready_chunks(automatic_amplification())
.widen_then(automatic_width(), |chunk| {
let keys = chunk
.iter()
.map(ser::serialize_to::<KeyBuf, _>)
.map(|result| result.expect("failed to serialize query key"))
.collect();
keys.ready_chunks(automatic_amplification())
.widen_then(automatic_width(), |chunk| {
let keys = chunk
.iter()
.map(ser::serialize_to::<KeyBuf, _>)
.map(|result| result.expect("failed to serialize query key"))
.collect();
self.db
.pool
.execute_get(Get { map: self.clone(), key: keys, res: None })
})
.map_ok(|results| results.into_iter().stream())
.try_flatten()
self.db
.pool
.execute_get(Get { map: self.clone(), key: keys, res: None })
})
.map_ok(|results| results.into_iter().stream())
.try_flatten()
}
}
src/database/map/remove.rs
+41 -43
View File
@@ -1,55 +1,53 @@
use std::{convert::AsRef, fmt::Debug, io::Write};
use conduwuit::{arrayvec::ArrayVec, implement};
use conduwuit::arrayvec::ArrayVec;
use serde::Serialize;
use crate::{keyval::KeyBuf, ser, util::or_else};
#[implement(super::Map)]
#[inline]
pub fn del<K>(&self, key: K)
where
K: Serialize + Debug,
{
let mut buf = KeyBuf::new();
self.bdel(key, &mut buf);
}
impl super::Map {
#[inline]
pub fn del<K>(&self, key: K)
where
K: Serialize + Debug,
{
let mut buf = KeyBuf::new();
self.bdel(key, &mut buf);
}
#[implement(super::Map)]
#[inline]
pub fn adel<const MAX: usize, K>(&self, key: K)
where
K: Serialize + Debug,
{
let mut buf = ArrayVec::<u8, MAX>::new();
self.bdel(key, &mut buf);
}
#[inline]
pub fn adel<const MAX: usize, K>(&self, key: K)
where
K: Serialize + Debug,
{
let mut buf = ArrayVec::<u8, MAX>::new();
self.bdel(key, &mut buf);
}
#[implement(super::Map)]
#[tracing::instrument(skip(self, buf), level = "trace")]
pub fn bdel<K, B>(&self, key: K, buf: &mut B)
where
K: Serialize + Debug,
B: Write + AsRef<[u8]>,
{
let key = ser::serialize(buf, key).expect("failed to serialize deletion key");
self.remove(key);
}
#[tracing::instrument(skip(self, buf), level = "trace")]
pub fn bdel<K, B>(&self, key: K, buf: &mut B)
where
K: Serialize + Debug,
B: Write + AsRef<[u8]>,
{
let key = ser::serialize(buf, key).expect("failed to serialize deletion key");
self.remove(key);
}
#[implement(super::Map)]
#[tracing::instrument(skip(self, key), fields(%self), level = "trace")]
pub fn remove<K>(&self, key: &K)
where
K: AsRef<[u8]> + ?Sized + Debug,
{
let write_options = &self.write_options;
self.db
.db
.delete_cf_opt(&self.cf(), key, write_options)
.or_else(or_else)
.expect("database remove error");
#[tracing::instrument(skip(self, key), fields(%self), level = "trace")]
pub fn remove<K>(&self, key: &K)
where
K: AsRef<[u8]> + ?Sized + Debug,
{
let write_options = &self.write_options;
self.db
.db
.delete_cf_opt(&self.cf(), key, write_options)
.or_else(or_else)
.expect("database remove error");
if !self.db.corked() {
self.db.flush().expect("database flush error");
if !self.db.corked() {
self.db.flush().expect("database flush error");
}
}
}
src/database/map/rev_keys.rs
+37 -37
View File
@@ -1,6 +1,6 @@
use std::sync::Arc;
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
use rocksdb::Direction;
use serde::Deserialize;
@@ -9,43 +9,43 @@
use super::rev_stream::is_cached;
use crate::{keyval, keyval::Key, stream};
#[implement(super::Map)]
pub fn rev_keys<'a, K>(self: &'a Arc<Self>) -> impl Stream<Item = Result<Key<'a, K>>> + Send
where
K: Deserialize<'a> + Send,
{
self.rev_raw_keys().map(keyval::result_deserialize_key::<K>)
}
#[implement(super::Map)]
#[tracing::instrument(skip(self), fields(%self), level = "trace")]
pub fn rev_raw_keys(self: &Arc<Self>) -> impl Stream<Item = Result<Key<'_>>> + Send {
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self) {
let state = state.init_rev(None);
return task::consume_budget()
.map(move |()| stream::KeysRev::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
impl super::Map {
pub fn rev_keys<'a, K>(self: &'a Arc<Self>) -> impl Stream<Item = Result<Key<'a, K>>> + Send
where
K: Deserialize<'a> + Send,
{
self.rev_raw_keys().map(keyval::result_deserialize_key::<K>)
}
let seek = Seek {
map: self.clone(),
dir: Direction::Reverse,
state: crate::pool::into_send_seek(state),
key: None,
res: None,
};
#[tracing::instrument(skip(self), fields(%self), level = "trace")]
pub fn rev_raw_keys(self: &Arc<Self>) -> impl Stream<Item = Result<Key<'_>>> + Send {
use crate::pool::Seek;
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::KeysRev<'_>>()
.into_stream()
.try_flatten()
.boxed()
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self) {
let state = state.init_rev(None);
return task::consume_budget()
.map(move |()| stream::KeysRev::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
}
let seek = Seek {
map: self.clone(),
dir: Direction::Reverse,
state: crate::pool::into_send_seek(state),
key: None,
res: None,
};
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::KeysRev<'_>>()
.into_stream()
.try_flatten()
.boxed()
}
}
src/database/map/rev_keys_from.rs
+67 -69
View File
@@ -1,6 +1,6 @@
use std::{convert::AsRef, fmt::Debug, sync::Arc};
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
use rocksdb::Direction;
use serde::{Deserialize, Serialize};
@@ -11,75 +11,73 @@
stream,
};
#[implement(super::Map)]
pub fn rev_keys_from<'a, K, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
{
self.rev_keys_from_raw(from)
.map(result_deserialize_key::<K>)
}
#[implement(super::Map)]
#[tracing::instrument(skip(self), level = "trace")]
pub fn rev_keys_from_raw<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(from).expect("failed to serialize query key");
self.rev_raw_keys_from(&key)
}
#[implement(super::Map)]
pub fn rev_keys_raw_from<'a, K, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: AsRef<[u8]> + ?Sized + Debug + Sync,
K: Deserialize<'a> + Send,
{
self.rev_raw_keys_from(from)
.map(result_deserialize_key::<K>)
}
#[implement(super::Map)]
#[tracing::instrument(skip(self, from), fields(%self), level = "trace")]
pub fn rev_raw_keys_from<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: AsRef<[u8]> + ?Sized + Debug,
{
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self, from) {
return stream::KeysRev::<'_>::from(state.init_rev(from.as_ref().into())).boxed();
impl super::Map {
pub fn rev_keys_from<'a, K, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
{
self.rev_keys_from_raw(from)
.map(result_deserialize_key::<K>)
}
let seek = Seek {
map: self.clone(),
dir: Direction::Reverse,
key: Some(from.as_ref().into()),
state: crate::pool::into_send_seek(state),
res: None,
};
#[tracing::instrument(skip(self), level = "trace")]
pub fn rev_keys_from_raw<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(from).expect("failed to serialize query key");
self.rev_raw_keys_from(&key)
}
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::KeysRev<'_>>()
.into_stream()
.try_flatten()
.boxed()
pub fn rev_keys_raw_from<'a, K, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: AsRef<[u8]> + ?Sized + Debug + Sync,
K: Deserialize<'a> + Send,
{
self.rev_raw_keys_from(from)
.map(result_deserialize_key::<K>)
}
#[tracing::instrument(skip(self, from), fields(%self), level = "trace")]
pub fn rev_raw_keys_from<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: AsRef<[u8]> + ?Sized + Debug,
{
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self, from) {
return stream::KeysRev::<'_>::from(state.init_rev(from.as_ref().into())).boxed();
}
let seek = Seek {
map: self.clone(),
dir: Direction::Reverse,
key: Some(from.as_ref().into()),
state: crate::pool::into_send_seek(state),
res: None,
};
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::KeysRev<'_>>()
.into_stream()
.try_flatten()
.boxed()
}
}
src/database/map/rev_keys_prefix.rs
+45 -47
View File
@@ -6,54 +6,52 @@
use crate::keyval::{Key, result_deserialize_key, serialize_key};
#[implement(super::Map)]
pub fn rev_keys_prefix<'a, K, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
{
self.rev_keys_prefix_raw(prefix)
.map(result_deserialize_key::<K>)
}
impl super::Map {
pub fn rev_keys_prefix<'a, K, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + use<'a, K, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
{
self.rev_keys_prefix_raw(prefix)
.map(result_deserialize_key::<K>)
}
#[implement(super::Map)]
#[tracing::instrument(skip(self), level = "trace")]
pub fn rev_keys_prefix_raw<P>(
self: &Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(prefix).expect("failed to serialize query key");
self.rev_raw_keys_from(&key)
.try_take_while(move |k: &Key<'_>| future::ok(k.starts_with(&key)))
}
#[tracing::instrument(skip(self), level = "trace")]
pub fn rev_keys_prefix_raw<P>(
self: &Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<Key<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(prefix).expect("failed to serialize query key");
self.rev_raw_keys_from(&key)
.try_take_while(move |k: &Key<'_>| future::ok(k.starts_with(&key)))
}
#[implement(super::Map)]
pub fn rev_keys_raw_prefix<'a, K, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
K: Deserialize<'a> + Send + 'a,
{
self.rev_raw_keys_prefix(prefix)
.map(result_deserialize_key::<K>)
}
pub fn rev_keys_raw_prefix<'a, K, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<Key<'a, K>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
K: Deserialize<'a> + Send + 'a,
{
self.rev_raw_keys_prefix(prefix)
.map(result_deserialize_key::<K>)
}
#[implement(super::Map)]
pub fn rev_raw_keys_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<Key<'a>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.rev_raw_keys_from(prefix)
.try_take_while(|k: &Key<'_>| future::ok(k.starts_with(prefix.as_ref())))
pub fn rev_raw_keys_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<Key<'a>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.rev_raw_keys_from(prefix)
.try_take_while(|k: &Key<'_>| future::ok(k.starts_with(prefix.as_ref())))
}
}
src/database/map/rev_stream.rs
+47 -47
View File
@@ -1,6 +1,6 @@
use std::sync::Arc;
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
use rocksdb::Direction;
use serde::Deserialize;
@@ -8,55 +8,55 @@
use crate::{keyval, keyval::KeyVal, stream};
/// Iterate key-value entries in the map from the end.
///
/// - Result is deserialized
#[implement(super::Map)]
pub fn rev_stream<'a, K, V>(
self: &'a Arc<Self>,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send
where
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.rev_raw_stream()
.map(keyval::result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map from the end.
///
/// - Result is raw
#[implement(super::Map)]
#[tracing::instrument(skip(self), fields(%self), level = "trace")]
pub fn rev_raw_stream(self: &Arc<Self>) -> impl Stream<Item = Result<KeyVal<'_>>> + Send {
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self) {
let state = state.init_rev(None);
return task::consume_budget()
.map(move |()| stream::ItemsRev::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
impl super::Map {
/// Iterate key-value entries in the map from the end.
///
/// - Result is deserialized
pub fn rev_stream<'a, K, V>(
self: &'a Arc<Self>,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send
where
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.rev_raw_stream()
.map(keyval::result_deserialize::<K, V>)
}
let seek = Seek {
map: self.clone(),
dir: Direction::Reverse,
state: crate::pool::into_send_seek(state),
key: None,
res: None,
};
/// Iterate key-value entries in the map from the end.
///
/// - Result is raw
#[tracing::instrument(skip(self), fields(%self), level = "trace")]
pub fn rev_raw_stream(self: &Arc<Self>) -> impl Stream<Item = Result<KeyVal<'_>>> + Send {
use crate::pool::Seek;
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::ItemsRev<'_>>()
.into_stream()
.try_flatten()
.boxed()
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self) {
let state = state.init_rev(None);
return task::consume_budget()
.map(move |()| stream::ItemsRev::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
}
let seek = Seek {
map: self.clone(),
dir: Direction::Reverse,
state: crate::pool::into_send_seek(state),
key: None,
res: None,
};
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::ItemsRev<'_>>()
.into_stream()
.try_flatten()
.boxed()
}
}
#[tracing::instrument(
src/database/map/rev_stream_from.rs
+90 -92
View File
@@ -1,6 +1,6 @@
use std::{convert::AsRef, fmt::Debug, sync::Arc};
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
use rocksdb::Direction;
use serde::{Deserialize, Serialize};
@@ -12,100 +12,98 @@
util::is_incomplete,
};
/// Iterate key-value entries in the map starting from upper-bound.
///
/// - Query is serialized
/// - Result is deserialized
#[implement(super::Map)]
pub fn rev_stream_from<'a, K, V, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.rev_stream_from_raw(from)
.map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map starting from upper-bound.
///
/// - Query is serialized
/// - Result is raw
#[implement(super::Map)]
#[tracing::instrument(skip(self), level = "trace")]
pub fn rev_stream_from_raw<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(from).expect("failed to serialize query key");
self.rev_raw_stream_from(&key)
}
/// Iterate key-value entries in the map starting from upper-bound.
///
/// - Query is raw
/// - Result is deserialized
#[implement(super::Map)]
pub fn rev_stream_raw_from<'a, K, V, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: AsRef<[u8]> + ?Sized + Debug + Sync,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.rev_raw_stream_from(from)
.map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map starting from upper-bound.
///
/// - Query is raw
/// - Result is raw
#[implement(super::Map)]
#[tracing::instrument(skip(self, from), fields(%self), level = "trace")]
pub fn rev_raw_stream_from<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: AsRef<[u8]> + ?Sized + Debug,
{
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self, from) {
let state = state.init_rev(from.as_ref().into());
return task::consume_budget()
.map(move |()| stream::ItemsRev::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
impl super::Map {
/// Iterate key-value entries in the map starting from upper-bound.
///
/// - Query is serialized
/// - Result is deserialized
pub fn rev_stream_from<'a, K, V, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.rev_stream_from_raw(from)
.map(result_deserialize::<K, V>)
}
let seek = Seek {
map: self.clone(),
dir: Direction::Reverse,
key: Some(from.as_ref().into()),
state: crate::pool::into_send_seek(state),
res: None,
};
/// Iterate key-value entries in the map starting from upper-bound.
///
/// - Query is serialized
/// - Result is raw
#[tracing::instrument(skip(self), level = "trace")]
pub fn rev_stream_from_raw<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(from).expect("failed to serialize query key");
self.rev_raw_stream_from(&key)
}
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::ItemsRev<'_>>()
.into_stream()
.try_flatten()
.boxed()
/// Iterate key-value entries in the map starting from upper-bound.
///
/// - Query is raw
/// - Result is deserialized
pub fn rev_stream_raw_from<'a, K, V, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: AsRef<[u8]> + ?Sized + Debug + Sync,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.rev_raw_stream_from(from)
.map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map starting from upper-bound.
///
/// - Query is raw
/// - Result is raw
#[tracing::instrument(skip(self, from), fields(%self), level = "trace")]
pub fn rev_raw_stream_from<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: AsRef<[u8]> + ?Sized + Debug,
{
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self, from) {
let state = state.init_rev(from.as_ref().into());
return task::consume_budget()
.map(move |()| stream::ItemsRev::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
}
let seek = Seek {
map: self.clone(),
dir: Direction::Reverse,
key: Some(from.as_ref().into()),
state: crate::pool::into_send_seek(state),
res: None,
};
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::ItemsRev<'_>>()
.into_stream()
.try_flatten()
.boxed()
}
}
#[tracing::instrument(
src/database/map/rev_stream_prefix.rs
+64 -66
View File
@@ -1,77 +1,75 @@
use std::{convert::AsRef, fmt::Debug, sync::Arc};
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{Stream, StreamExt, TryStreamExt, future};
use serde::{Deserialize, Serialize};
use crate::keyval::{KeyVal, result_deserialize, serialize_key};
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is serialized
/// - Result is deserialized
#[implement(super::Map)]
pub fn rev_stream_prefix<'a, K, V, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.rev_stream_prefix_raw(prefix)
.map(result_deserialize::<K, V>)
}
impl super::Map {
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is serialized
/// - Result is deserialized
pub fn rev_stream_prefix<'a, K, V, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.rev_stream_prefix_raw(prefix)
.map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is serialized
/// - Result is raw
#[implement(super::Map)]
#[tracing::instrument(skip(self), level = "trace")]
pub fn rev_stream_prefix_raw<P>(
self: &Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(prefix).expect("failed to serialize query key");
self.rev_raw_stream_from(&key)
.try_take_while(move |(k, _): &KeyVal<'_>| future::ok(k.starts_with(&key)))
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is serialized
/// - Result is raw
#[tracing::instrument(skip(self), level = "trace")]
pub fn rev_stream_prefix_raw<P>(
self: &Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(prefix).expect("failed to serialize query key");
self.rev_raw_stream_from(&key)
.try_take_while(move |(k, _): &KeyVal<'_>| future::ok(k.starts_with(&key)))
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is raw
/// - Result is deserialized
#[implement(super::Map)]
pub fn rev_stream_raw_prefix<'a, K, V, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
K: Deserialize<'a> + Send + 'a,
V: Deserialize<'a> + Send + 'a,
{
self.rev_raw_stream_prefix(prefix)
.map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is raw
/// - Result is deserialized
pub fn rev_stream_raw_prefix<'a, K, V, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
K: Deserialize<'a> + Send + 'a,
V: Deserialize<'a> + Send + 'a,
{
self.rev_raw_stream_prefix(prefix)
.map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is raw
/// - Result is raw
#[implement(super::Map)]
pub fn rev_raw_stream_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<KeyVal<'a>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.rev_raw_stream_from(prefix)
.try_take_while(|(k, _): &KeyVal<'_>| future::ok(k.starts_with(prefix.as_ref())))
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is raw
/// - Result is raw
pub fn rev_raw_stream_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<KeyVal<'a>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.rev_raw_stream_from(prefix)
.try_take_while(|(k, _): &KeyVal<'_>| future::ok(k.starts_with(prefix.as_ref())))
}
}
src/database/map/stream.rs
+46 -46
View File
@@ -1,6 +1,6 @@
use std::sync::Arc;
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
use rocksdb::Direction;
use serde::Deserialize;
@@ -8,54 +8,54 @@
use crate::{keyval, keyval::KeyVal, stream};
/// Iterate key-value entries in the map from the beginning.
///
/// - Result is deserialized
#[implement(super::Map)]
pub fn stream<'a, K, V>(
self: &'a Arc<Self>,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send
where
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.raw_stream().map(keyval::result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map from the beginning.
///
/// - Result is raw
#[implement(super::Map)]
#[tracing::instrument(skip(self), fields(%self), level = "trace")]
pub fn raw_stream(self: &Arc<Self>) -> impl Stream<Item = Result<KeyVal<'_>>> + Send {
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self) {
let state = state.init_fwd(None);
return task::consume_budget()
.map(move |()| stream::Items::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
impl super::Map {
/// Iterate key-value entries in the map from the beginning.
///
/// - Result is deserialized
pub fn stream<'a, K, V>(
self: &'a Arc<Self>,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send
where
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.raw_stream().map(keyval::result_deserialize::<K, V>)
}
let seek = Seek {
map: self.clone(),
dir: Direction::Forward,
state: crate::pool::into_send_seek(state),
key: None,
res: None,
};
/// Iterate key-value entries in the map from the beginning.
///
/// - Result is raw
#[tracing::instrument(skip(self), fields(%self), level = "trace")]
pub fn raw_stream(self: &Arc<Self>) -> impl Stream<Item = Result<KeyVal<'_>>> + Send {
use crate::pool::Seek;
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::Items<'_>>()
.into_stream()
.try_flatten()
.boxed()
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self) {
let state = state.init_fwd(None);
return task::consume_budget()
.map(move |()| stream::Items::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
}
let seek = Seek {
map: self.clone(),
dir: Direction::Forward,
state: crate::pool::into_send_seek(state),
key: None,
res: None,
};
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::Items<'_>>()
.into_stream()
.try_flatten()
.boxed()
}
}
#[tracing::instrument(
src/database/map/stream_from.rs
+92 -94
View File
@@ -1,6 +1,6 @@
use std::{convert::AsRef, fmt::Debug, sync::Arc};
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
use rocksdb::Direction;
use serde::{Deserialize, Serialize};
@@ -11,105 +11,103 @@
stream,
};
/// Iterate key-value entries in the map starting from lower-bound.
///
/// - Query is serialized
/// - Result is deserialized
#[implement(super::Map)]
pub fn stream_from<'a, K, V, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.stream_from_raw(from).map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map starting from lower-bound.
///
/// - Query is serialized
/// - Result is raw
#[implement(super::Map)]
#[tracing::instrument(skip(self), level = "trace")]
pub fn stream_from_raw<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(from).expect("failed to serialize query key");
self.raw_stream_from(&key)
}
/// Iterate key-value entries in the map starting from lower-bound.
///
/// - Query is raw
/// - Result is deserialized
#[implement(super::Map)]
pub fn stream_raw_from<'a, K, V, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: AsRef<[u8]> + ?Sized + Debug + Sync,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.raw_stream_from(from).map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map starting from lower-bound.
///
/// - Query is raw
/// - Result is raw
#[implement(super::Map)]
#[tracing::instrument(skip(self, from), fields(%self), level = "trace")]
pub fn raw_stream_from<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: AsRef<[u8]> + ?Sized + Debug,
{
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self, from) {
let state = state.init_fwd(from.as_ref().into());
return task::consume_budget()
.map(move |()| stream::Items::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
impl super::Map {
/// Iterate key-value entries in the map starting from lower-bound.
///
/// - Query is serialized
/// - Result is deserialized
pub fn stream_from<'a, K, V, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.stream_from_raw(from).map(result_deserialize::<K, V>)
}
let seek = Seek {
map: self.clone(),
dir: Direction::Forward,
key: Some(from.as_ref().into()),
state: crate::pool::into_send_seek(state),
res: None,
};
/// Iterate key-value entries in the map starting from lower-bound.
///
/// - Query is serialized
/// - Result is raw
#[tracing::instrument(skip(self), level = "trace")]
pub fn stream_from_raw<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(from).expect("failed to serialize query key");
self.raw_stream_from(&key)
}
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::Items<'_>>()
.into_stream()
.try_flatten()
.boxed()
/// Iterate key-value entries in the map starting from lower-bound.
///
/// - Query is raw
/// - Result is deserialized
pub fn stream_raw_from<'a, K, V, P>(
self: &'a Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: AsRef<[u8]> + ?Sized + Debug + Sync,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.raw_stream_from(from).map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map starting from lower-bound.
///
/// - Query is raw
/// - Result is raw
#[tracing::instrument(skip(self, from), fields(%self), level = "trace")]
pub fn raw_stream_from<P>(
self: &Arc<Self>,
from: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: AsRef<[u8]> + ?Sized + Debug,
{
use crate::pool::Seek;
let opts = super::iter_options_default(&self.db);
let state = stream::State::new(self, opts);
if is_cached(self, from) {
let state = state.init_fwd(from.as_ref().into());
return task::consume_budget()
.map(move |()| stream::Items::<'_>::from(state))
.into_stream()
.flatten()
.boxed();
}
let seek = Seek {
map: self.clone(),
dir: Direction::Forward,
key: Some(from.as_ref().into()),
state: crate::pool::into_send_seek(state),
res: None,
};
self.db
.pool
.execute_iter(seek)
.ok_into::<stream::Items<'_>>()
.into_stream()
.try_flatten()
.boxed()
}
}
#[tracing::instrument(
name = "cached",
level = "trace",
skip(map, from),
fields(%map),
name = "cached",
level = "trace",
skip(map, from),
fields(%map),
)]
pub(super) fn is_cached<P>(map: &Arc<super::Map>, from: &P) -> bool
where
src/database/map/stream_prefix.rs
+64 -66
View File
@@ -1,77 +1,75 @@
use std::{convert::AsRef, fmt::Debug, sync::Arc};
use conduwuit::{Result, implement};
use conduwuit::Result;
use futures::{Stream, StreamExt, TryStreamExt, future};
use serde::{Deserialize, Serialize};
use crate::keyval::{KeyVal, result_deserialize, serialize_key};
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is serialized
/// - Result is deserialized
#[implement(super::Map)]
pub fn stream_prefix<'a, K, V, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.stream_prefix_raw(prefix)
.map(result_deserialize::<K, V>)
}
impl super::Map {
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is serialized
/// - Result is deserialized
pub fn stream_prefix<'a, K, V, P>(
self: &'a Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + use<'a, K, V, P>
where
P: Serialize + ?Sized + Debug,
K: Deserialize<'a> + Send,
V: Deserialize<'a> + Send,
{
self.stream_prefix_raw(prefix)
.map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is serialized
/// - Result is raw
#[implement(super::Map)]
#[tracing::instrument(skip(self), level = "trace")]
pub fn stream_prefix_raw<P>(
self: &Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(prefix).expect("failed to serialize query key");
self.raw_stream_from(&key)
.try_take_while(move |(k, _): &KeyVal<'_>| future::ok(k.starts_with(&key)))
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is serialized
/// - Result is raw
#[tracing::instrument(skip(self), level = "trace")]
pub fn stream_prefix_raw<P>(
self: &Arc<Self>,
prefix: &P,
) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
where
P: Serialize + ?Sized + Debug,
{
let key = serialize_key(prefix).expect("failed to serialize query key");
self.raw_stream_from(&key)
.try_take_while(move |(k, _): &KeyVal<'_>| future::ok(k.starts_with(&key)))
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is raw
/// - Result is deserialized
#[implement(super::Map)]
pub fn stream_raw_prefix<'a, K, V, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
K: Deserialize<'a> + Send + 'a,
V: Deserialize<'a> + Send + 'a,
{
self.raw_stream_prefix(prefix)
.map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is raw
/// - Result is deserialized
pub fn stream_raw_prefix<'a, K, V, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<KeyVal<'a, K, V>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
K: Deserialize<'a> + Send + 'a,
V: Deserialize<'a> + Send + 'a,
{
self.raw_stream_prefix(prefix)
.map(result_deserialize::<K, V>)
}
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is raw
/// - Result is raw
#[implement(super::Map)]
pub fn raw_stream_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<KeyVal<'a>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.raw_stream_from(prefix)
.try_take_while(|(k, _): &KeyVal<'_>| future::ok(k.starts_with(prefix.as_ref())))
/// Iterate key-value entries in the map where the key matches a prefix.
///
/// - Query is raw
/// - Result is raw
pub fn raw_stream_prefix<'a, P>(
self: &'a Arc<Self>,
prefix: &'a P,
) -> impl Stream<Item = Result<KeyVal<'a>>> + Send + 'a
where
P: AsRef<[u8]> + ?Sized + Debug + Sync + 'a,
{
self.raw_stream_from(prefix)
.try_take_while(|(k, _): &KeyVal<'_>| future::ok(k.starts_with(prefix.as_ref())))
}
}