xs/nu/commands/
cat_stream_command.rs

1use nu_engine::CallExt;
2use nu_protocol::engine::{Call, Command, EngineState, Stack};
3use nu_protocol::{
4    Category, ListStream, PipelineData, ShellError, Signals, Signature, SyntaxShape, Type, Value,
5};
6use std::time::Duration;
7
8use crate::store::{FollowOption, ReadOptions, Store};
9
10#[derive(Clone)]
11pub struct CatStreamCommand {
12    store: Store,
13}
14
15impl CatStreamCommand {
16    pub fn new(store: Store) -> Self {
17        Self { store }
18    }
19}
20
21impl Command for CatStreamCommand {
22    fn name(&self) -> &str {
23        ".cat"
24    }
25
26    fn signature(&self) -> Signature {
27        Signature::build(".cat")
28            .input_output_types(vec![(Type::Nothing, Type::Any)])
29            .switch("follow", "long poll for new events", Some('f'))
30            .named(
31                "pulse",
32                SyntaxShape::Int,
33                "interval in ms for synthetic xs.pulse events",
34                Some('p'),
35            )
36            .switch("new", "skip existing, only show new", Some('n'))
37            .switch("detail", "include all frame fields", Some('d'))
38            .named(
39                "limit",
40                SyntaxShape::Int,
41                "limit the number of frames to retrieve",
42                None,
43            )
44            .named(
45                "after",
46                SyntaxShape::String,
47                "start after a specific frame ID (exclusive)",
48                Some('a'),
49            )
50            .named("topic", SyntaxShape::String, "filter by topic", Some('T'))
51            .category(Category::Experimental)
52    }
53
54    fn description(&self) -> &str {
55        "Reads the event stream and returns frames (streaming version)"
56    }
57
58    fn run(
59        &self,
60        engine_state: &EngineState,
61        stack: &mut Stack,
62        call: &Call,
63        _input: PipelineData,
64    ) -> Result<PipelineData, ShellError> {
65        let follow = call.has_flag(engine_state, stack, "follow")?;
66        let pulse: Option<i64> = call.get_flag(engine_state, stack, "pulse")?;
67        let new = call.has_flag(engine_state, stack, "new")?;
68        let detail = call.has_flag(engine_state, stack, "detail")?;
69        let limit: Option<i64> = call.get_flag(engine_state, stack, "limit")?;
70        let after: Option<String> = call.get_flag(engine_state, stack, "after")?;
71        let topic: Option<String> = call.get_flag(engine_state, stack, "topic")?;
72
73        // Parse after
74        let after: Option<scru128::Scru128Id> = after
75            .as_deref()
76            .map(|s| {
77                s.parse().map_err(|e| ShellError::GenericError {
78                    error: "Invalid after".into(),
79                    msg: format!("Failed to parse Scru128Id: {e}"),
80                    span: Some(call.head),
81                    help: None,
82                    inner: vec![],
83                })
84            })
85            .transpose()?;
86
87        // Build ReadOptions
88        let options = ReadOptions::builder()
89            .follow(if let Some(pulse_ms) = pulse {
90                FollowOption::WithHeartbeat(Duration::from_millis(pulse_ms as u64))
91            } else if follow {
92                FollowOption::On
93            } else {
94                FollowOption::Off
95            })
96            .new(new)
97            .maybe_after(after)
98            .maybe_limit(limit.map(|l| l as usize))
99            .maybe_topic(topic.clone())
100            .build();
101
102        let store = self.store.clone();
103        let span = call.head;
104        let signals = engine_state.signals().clone();
105
106        // Create channel for async -> sync bridge
107        let (tx, rx) = std::sync::mpsc::channel();
108
109        // Spawn thread to handle async store.read()
110        std::thread::spawn(move || {
111            let rt = tokio::runtime::Runtime::new().expect("Failed to create tokio runtime");
112            rt.block_on(async move {
113                let mut receiver = store.read(options).await;
114
115                while let Some(frame) = receiver.recv().await {
116                    // Convert frame to Nu value
117                    let mut value = crate::nu::util::frame_to_value(&frame, span);
118
119                    // Filter fields if not --detail
120                    if !detail {
121                        value = match value {
122                            Value::Record { val, .. } => {
123                                let mut filtered = val.into_owned();
124                                filtered.remove("ttl");
125                                Value::record(filtered, span)
126                            }
127                            v => v,
128                        };
129                    }
130
131                    if tx.send(value).is_err() {
132                        break;
133                    }
134                }
135            });
136        });
137
138        // Create ListStream from channel with signal-aware polling
139        let stream = ListStream::new(
140            std::iter::from_fn(move || {
141                use std::sync::mpsc::RecvTimeoutError;
142                loop {
143                    if signals.interrupted() {
144                        return None;
145                    }
146                    match rx.recv_timeout(Duration::from_millis(100)) {
147                        Ok(value) => return Some(value),
148                        Err(RecvTimeoutError::Timeout) => continue,
149                        Err(RecvTimeoutError::Disconnected) => return None,
150                    }
151                }
152            }),
153            span,
154            Signals::empty(),
155        );
156
157        Ok(PipelineData::ListStream(stream, None))
158    }
159}
xs/nu/commands/cat_stream_command.rs

xs/nu/commands/
cat_stream_command.rs
