use std::{convert::AsRef, fmt::Debug, sync::Arc}; use conduwuit::{Result, implement}; use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt}; use rocksdb::Direction; use serde::{Deserialize, Serialize}; use tokio::task; use crate::{ keyval::{KeyVal, result_deserialize, serialize_key}, stream, 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, from: &P, ) -> impl Stream>> + 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::) } /// 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

( self: &Arc, from: &P, ) -> impl Stream>> + 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, from: &P, ) -> impl Stream>> + 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::) } /// 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

( self: &Arc, from: &P, ) -> impl Stream>> + 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::>() .into_stream() .try_flatten() .boxed() } #[tracing::instrument( name = "cached", level = "trace", skip(map, from), fields(%map), )] pub(super) fn is_cached

(map: &Arc, from: &P) -> bool where P: AsRef<[u8]> + ?Sized, { let cache_opts = super::cache_iter_options_default(&map.db); let cache_status = stream::State::new(map, cache_opts) .init_rev(from.as_ref().into()) .status(); !matches!(cache_status, Some(e) if is_incomplete(&e)) }