tuwunel_database/map/
rev_stream_from.rs

1use std::{convert::AsRef, fmt::Debug, sync::Arc};
2
3use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt, future::Either};
4use rocksdb::Direction;
5use serde::{Deserialize, Serialize};
6use tokio::task;
7use tuwunel_core::{Result, implement};
8
9use crate::{
10	keyval::{KeyVal, result_deserialize, serialize_key},
11	stream,
12	util::is_incomplete,
13};
14
15/// Iterate key-value entries in the map starting from upper-bound.
16///
17/// - Query is serialized
18/// - Result is deserialized
19#[implement(super::Map)]
20pub fn rev_stream_from<'a, K, V, P>(
21	self: &'a Arc<Self>,
22	from: &P,
23) -> impl Stream<Item = Result<KeyVal<'_, K, V>>> + Send + use<'a, K, V, P>
24where
25	P: Serialize + ?Sized + Debug,
26	K: Deserialize<'a> + Send,
27	V: Deserialize<'a> + Send,
28{
29	self.rev_stream_from_raw(from)
30		.map(result_deserialize::<K, V>)
31}
32
33/// Iterate key-value entries in the map starting from upper-bound.
34///
35/// - Query is serialized
36/// - Result is raw
37#[implement(super::Map)]
38#[tracing::instrument(skip(self), level = "trace")]
39pub fn rev_stream_from_raw<P>(
40	self: &Arc<Self>,
41	from: &P,
42) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
43where
44	P: Serialize + ?Sized + Debug,
45{
46	let key = serialize_key(from).expect("failed to serialize query key");
47	self.rev_raw_stream_from(&key)
48}
49
50/// Iterate key-value entries in the map starting from upper-bound.
51///
52/// - Query is raw
53/// - Result is deserialized
54#[implement(super::Map)]
55pub fn rev_stream_raw_from<'a, K, V, P>(
56	self: &'a Arc<Self>,
57	from: &P,
58) -> impl Stream<Item = Result<KeyVal<'_, K, V>>> + Send + use<'a, K, V, P>
59where
60	P: AsRef<[u8]> + ?Sized + Debug + Sync,
61	K: Deserialize<'a> + Send,
62	V: Deserialize<'a> + Send,
63{
64	self.rev_raw_stream_from(from)
65		.map(result_deserialize::<K, V>)
66}
67
68/// Iterate key-value entries in the map starting from upper-bound.
69///
70/// - Query is raw
71/// - Result is raw
72#[implement(super::Map)]
73#[tracing::instrument(skip(self, from), fields(%self), level = "trace")]
74pub fn rev_raw_stream_from<P>(
75	self: &Arc<Self>,
76	from: &P,
77) -> impl Stream<Item = Result<KeyVal<'_>>> + Send + use<'_, P>
78where
79	P: AsRef<[u8]> + ?Sized + Debug,
80{
81	use crate::pool::Seek;
82
83	let opts = super::iter_options_default(&self.engine);
84	let state = stream::State::new(self, opts);
85	if is_cached(self, from) {
86		let state = state.init_rev(from.as_ref().into());
87		return Either::Left(
88			task::consume_budget()
89				.map(move |()| stream::ItemsRev::<'_>::from(state))
90				.into_stream()
91				.flatten(),
92		);
93	}
94
95	let seek = Seek {
96		map: self.clone(),
97		dir: Direction::Reverse,
98		key: Some(from.as_ref().into()),
99		state: crate::pool::into_send_seek(state),
100		res: None,
101	};
102
103	Either::Right(
104		self.engine
105			.pool
106			.execute_iter(seek)
107			.ok_into::<stream::ItemsRev<'_>>()
108			.into_stream()
109			.try_flatten(),
110	)
111}
112
113#[tracing::instrument(
114    name = "cached",
115    level = "trace",
116    skip(map, from),
117    fields(%map),
118)]
119pub(super) fn is_cached<P>(map: &Arc<super::Map>, from: &P) -> bool
120where
121	P: AsRef<[u8]> + ?Sized,
122{
123	let cache_opts = super::cache_iter_options_default(&map.engine);
124	let cache_status = stream::State::new(map, cache_opts)
125		.init_rev(from.as_ref().into())
126		.status();
127
128	!matches!(cache_status, Some(e) if is_incomplete(&e))
129}
tuwunel_database/map/rev_stream_from.rs

tuwunel_database/map/
rev_stream_from.rs
