use anyhow::Result; use crossterm::event::{self, Event, KeyCode, KeyModifiers}; use std::{ sync::{ atomic::{AtomicBool, AtomicU64, Ordering}, Arc, }, time::Duration, }; pub type AbortSignal = Arc; #[derive(Debug, Default)] pub struct QueryIdGen { counter: AtomicU64, parents: parking_lot::RwLock>, } impl QueryIdGen { pub fn root_id(&self) -> u64 { self.counter.fetch_add(1, Ordering::SeqCst) } pub fn child_id(&self, parent_path: &[u64]) -> u64 { if parent_path.is_empty() { return self.root_id(); } let key = parent_path .iter() .map(|v| v.to_string()) .collect::>() .join("/"); let mut parents = self.parents.write(); let entry = parents.entry(key).or_insert(0); let id = *entry; *entry += 1; id } } pub fn format_query_path(path: &[u64]) -> String { if path.is_empty() { return String::new(); } format!( "[{}]", path.iter() .map(|v| v.to_string()) .collect::>() .join("/") ) } tokio::task_local! { static QUERY_SCOPE: String; static QUERY_ID_GEN: QueryIdGen; } pub fn current_query_scope() -> Option { QUERY_SCOPE .try_with(|v| if v.is_empty() { None } else { Some(v.clone()) }) .ok() .flatten() } pub async fn with_query_scope(scope: String, future: F) -> F::Output { QUERY_SCOPE.scope(scope, future).await } pub fn next_query_id(parent_path: &[u64]) -> u64 { if parent_path.is_empty() { return QUERY_ID_GEN.try_with(|g| g.root_id()).unwrap_or(0); } QUERY_ID_GEN .try_with(|g| g.child_id(parent_path)) .unwrap_or(0) } pub struct AbortSignalInner { ctrlc: AtomicBool, ctrld: AtomicBool, } pub fn create_abort_signal() -> AbortSignal { AbortSignalInner::new() } impl AbortSignalInner { pub fn new() -> AbortSignal { Arc::new(Self { ctrlc: AtomicBool::new(false), ctrld: AtomicBool::new(false), }) } pub fn aborted(&self) -> bool { if self.aborted_ctrlc() { return true; } if self.aborted_ctrld() { return true; } false } pub fn aborted_ctrlc(&self) -> bool { self.ctrlc.load(Ordering::SeqCst) } pub fn aborted_ctrld(&self) -> bool { self.ctrld.load(Ordering::SeqCst) } pub fn reset(&self) { self.ctrlc.store(false, Ordering::SeqCst); self.ctrld.store(false, Ordering::SeqCst); } pub fn set_ctrlc(&self) { self.ctrlc.store(true, Ordering::SeqCst); } pub fn set_ctrld(&self) { self.ctrld.store(true, Ordering::SeqCst); } } pub async fn wait_abort_signal(abort_signal: &AbortSignal) { loop { if abort_signal.aborted() { break; } tokio::time::sleep(std::time::Duration::from_millis(25)).await; } } pub fn poll_abort_signal(abort_signal: &AbortSignal) -> Result { if crossterm::event::poll(Duration::from_millis(25))? { if let Event::Key(key) = event::read()? { match key.code { KeyCode::Char('c') if key.modifiers == KeyModifiers::CONTROL => { abort_signal.set_ctrlc(); return Ok(true); } KeyCode::Char('d') if key.modifiers == KeyModifiers::CONTROL => { abort_signal.set_ctrld(); return Ok(true); } _ => {} } } } Ok(false) }