nu_plugin_skim 0.25.0

An `sk` command that can handle Nushell's structured data
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

use std::sync::Arc;

use nu_protocol::{PipelineData, Span, Spanned, Value, engine::Closure};
use shlex::Shlex;
use skim::reader::CommandCollector;
use skim::{SkimItem, prelude::unbounded};

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>) {
        let (tx, rx) = unbounded::<Arc<dyn SkimItem>>();
        let (tx_interrupt, mut rx_interrupt) = unbounded();
        let context = self.context.clone();
        let closure = self.closure.clone();
        let cmd = Shlex::new(cmd)
            .next()
            .expect("Skim's {q} should have produced a single shell-quoted value");
        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(cmd, Span::unknown())],
                PipelineData::Empty,
                true,
                true,
            ) {
                Ok(PipelineData::ByteStream(stream, _)) => {
                    let span = stream.span();
                    if let Some(lines) = stream.lines() {
                        for (index, line) in lines.enumerate() {
                            if rx_interrupt.try_recv().is_ok() {
                                break;
                            }
                            let send_result = match line {
                                Ok(line) => tx.send(Arc::new(NuItem::new(
                                    index,
                                    context.clone(),
                                    Value::string(line, span),
                                ))),
                                Err(err) => tx.send(Arc::new(NuItem::new(
                                    index,
                                    context.clone(),
                                    Value::error(err, span),
                                ))),
                            };
                            if send_result.is_err() {
                                break;
                            }
                        }
                    }
                }
                Ok(stream) => {
                    for (index, value) in stream.into_iter().enumerate() {
                        if rx_interrupt.try_recv().is_ok() {
                            break;
                        }
                        let send_result =
                            tx.send(Arc::new(NuItem::new(index, context.clone(), value)));
                        if send_result.is_err() {
                            break;
                        }
                    }
                }
                Err(err) => {
                    let _ = tx.send(Arc::new(NuItem::new(
                        0,
                        context.clone(),
                        Value::error(err, Span::unknown()),
                    )));
                }
            }

            components_to_stop.fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
        });
        (rx, tx_interrupt)
    }
}