1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
use std::sync::Arc;
use nu_protocol::{engine::Closure, PipelineData, Span, Spanned, Value};
use skim::{prelude::unbounded, CommandCollector, SkimItem};
use crate::{command_context::CommandContext, nu_item::NuItem};
pub struct NuCommandCollector {
pub context: Arc<CommandContext>,
pub closure: Spanned<Closure>,
}
impl CommandCollector for NuCommandCollector {
fn invoke(
&mut self,
cmd: &str, // not really the command - actually the query string
components_to_stop: std::sync::Arc<std::sync::atomic::AtomicUsize>,
) -> (
skim::SkimItemReceiver,
skim::prelude::Sender<i32>,
Option<std::thread::JoinHandle<()>>,
) {
let (tx, rx) = unbounded::<Arc<dyn SkimItem>>();
let (tx_interrupt, rx_interrupt) = unbounded();
let context = self.context.clone();
let closure = self.closure.clone();
let cmd = cmd.to_owned();
(
rx,
tx_interrupt,
Some(std::thread::spawn(move || {
components_to_stop.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
match context.engine.eval_closure_with_stream(
&closure,
vec![Value::String {
val: cmd,
internal_span: Span::unknown(),
}],
PipelineData::Empty,
true,
true,
) {
Ok(PipelineData::ByteStream(stream, _)) => {
let span = stream.span();
if let Some(lines) = stream.lines() {
for line in lines {
if rx_interrupt.try_recv().is_ok() {
break;
}
let send_result = match line {
Ok(line) => tx.try_send(Arc::new(NuItem {
context: context.clone(),
value: Value::string(line, span),
})),
Err(err) => tx.try_send(Arc::new(NuItem {
context: context.clone(),
value: Value::error(err, span),
})),
};
if send_result.is_err() {
break;
}
}
}
}
Ok(stream) => {
for value in stream {
if rx_interrupt.try_recv().is_ok() {
break;
}
let send_result = tx.try_send(Arc::new(NuItem {
context: context.clone(),
value,
}));
if send_result.is_err() {
break;
}
}
}
Err(err) => {
let _ = tx.try_send(Arc::new(NuItem {
context: context.clone(),
value: Value::error(err, Span::unknown()),
}));
}
}
components_to_stop.fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
})),
)
}
}