xs/nu/commands/
cat_stream_command.rs

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

xs/nu/commands/
cat_stream_command.rs
