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(
51                "from",
52                SyntaxShape::String,
53                "start from a specific frame ID (inclusive)",
54                None,
55            )
56            .named(
57                "last",
58                SyntaxShape::Int,
59                "return the N most recent frames",
60                None,
61            )
62            .named("topic", SyntaxShape::String, "filter by topic", Some('T'))
63            .category(Category::Experimental)
64    }
65
66    fn description(&self) -> &str {
67        "Reads the event stream and returns frames (streaming version)"
68    }
69
70    fn run(
71        &self,
72        engine_state: &EngineState,
73        stack: &mut Stack,
74        call: &Call,
75        _input: PipelineData,
76    ) -> Result<PipelineData, ShellError> {
77        let follow = call.has_flag(engine_state, stack, "follow")?;
78        let pulse: Option<i64> = call.get_flag(engine_state, stack, "pulse")?;
79        let new = call.has_flag(engine_state, stack, "new")?;
80        let detail = call.has_flag(engine_state, stack, "detail")?;
81        let limit: Option<i64> = call.get_flag(engine_state, stack, "limit")?;
82        let last: Option<i64> = call.get_flag(engine_state, stack, "last")?;
83        let after: Option<String> = call.get_flag(engine_state, stack, "after")?;
84        let from: Option<String> = call.get_flag(engine_state, stack, "from")?;
85        let topic: Option<String> = call.get_flag(engine_state, stack, "topic")?;
86
87        // Helper to parse Scru128Id
88        let parse_id = |s: &str, name: &str| -> Result<scru128::Scru128Id, ShellError> {
89            s.parse().map_err(|e| ShellError::GenericError {
90                error: format!("Invalid {name}"),
91                msg: format!("Failed to parse Scru128Id: {e}"),
92                span: Some(call.head),
93                help: None,
94                inner: vec![],
95            })
96        };
97
98        let after: Option<scru128::Scru128Id> =
99            after.as_deref().map(|s| parse_id(s, "after")).transpose()?;
100        let from: Option<scru128::Scru128Id> =
101            from.as_deref().map(|s| parse_id(s, "from")).transpose()?;
102
103        // Build ReadOptions
104        let options = ReadOptions::builder()
105            .follow(if let Some(pulse_ms) = pulse {
106                FollowOption::WithHeartbeat(Duration::from_millis(pulse_ms as u64))
107            } else if follow {
108                FollowOption::On
109            } else {
110                FollowOption::Off
111            })
112            .new(new)
113            .maybe_after(after)
114            .maybe_from(from)
115            .maybe_limit(limit.map(|l| l as usize))
116            .maybe_last(last.map(|l| l as usize))
117            .maybe_topic(topic.clone())
118            .build();
119
120        let store = self.store.clone();
121        let span = call.head;
122        let signals = engine_state.signals().clone();
123
124        // Create channel for async -> sync bridge
125        let (tx, rx) = std::sync::mpsc::channel();
126
127        // Spawn thread to handle async store.read()
128        std::thread::spawn(move || {
129            let rt = tokio::runtime::Runtime::new().expect("Failed to create tokio runtime");
130            rt.block_on(async move {
131                let mut receiver = store.read(options).await;
132
133                while let Some(frame) = receiver.recv().await {
134                    // Convert frame to Nu value
135                    let mut value = crate::nu::util::frame_to_value(&frame, span);
136
137                    // Filter fields if not --detail
138                    if !detail {
139                        value = match value {
140                            Value::Record { val, .. } => {
141                                let mut filtered = val.into_owned();
142                                filtered.remove("ttl");
143                                Value::record(filtered, span)
144                            }
145                            v => v,
146                        };
147                    }
148
149                    if tx.send(value).is_err() {
150                        break;
151                    }
152                }
153            });
154        });
155
156        // Create ListStream from channel with signal-aware polling
157        let stream = ListStream::new(
158            std::iter::from_fn(move || {
159                use std::sync::mpsc::RecvTimeoutError;
160                loop {
161                    if signals.interrupted() {
162                        return None;
163                    }
164                    match rx.recv_timeout(Duration::from_millis(100)) {
165                        Ok(value) => return Some(value),
166                        Err(RecvTimeoutError::Timeout) => continue,
167                        Err(RecvTimeoutError::Disconnected) => return None,
168                    }
169                }
170            }),
171            span,
172            Signals::empty(),
173        );
174
175        Ok(PipelineData::ListStream(stream, None))
176    }
177}
xs/nu/commands/cat_stream_command.rs

xs/nu/commands/
cat_stream_command.rs
