tuwunel_admin/query/raw/
iter.rs

1use futures::{FutureExt, StreamExt, TryStreamExt};
2use tokio::time::Instant;
3use tuwunel_core::{Result, apply, utils::TryReadyExt};
4use tuwunel_database::KeyVal;
5
6use super::{decode, encode};
7use crate::admin_command;
8
9#[admin_command]
10pub(super) async fn raw_iter(
11	&self,
12	map: String,
13	prefix: Option<String>,
14	limit: Option<usize>,
15	from: Option<String>,
16	backwards: bool,
17) -> Result {
18	writeln!(self, "```").await?;
19
20	let map = self.services.db.get(&map)?;
21	let timer = Instant::now();
22
23	let prefix = prefix.as_deref().map(decode);
24	let from = from.as_deref().map(decode);
25
26	let stream = match from.as_ref().or(prefix.as_ref()) {
27		| Some(from) if !backwards => map.raw_stream_from(from).boxed(),
28		| Some(from) => map.rev_raw_stream_from(from).boxed(),
29		| None if !backwards => map.raw_stream().boxed(),
30		| None => map.rev_raw_stream().boxed(),
31	};
32
33	stream
34		.ready_try_take_while(|(k, _): &KeyVal<'_>| {
35			Ok(prefix
36				.as_ref()
37				.map(|prefix| k.starts_with(prefix))
38				.unwrap_or(true))
39		})
40		.take(limit.unwrap_or(usize::MAX))
41		.map_ok(apply!(2, encode))
42		.try_for_each(|(key, val)| writeln!(self, "{key} => {val}"))
43		.boxed()
44		.await?;
45
46	let query_time = timer.elapsed();
47	write!(self, "\n```\n\nQuery completed in {query_time:?}").await
48}
tuwunel_admin/query/raw/iter.rs

tuwunel_admin/query/raw/
iter.rs
