cross-stream 0.12.0

An event stream store for personal, local-first use, specializing in event sourcing.
Documentation 
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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

use nu_engine::CallExt;
use nu_protocol::engine::{Call, Command, EngineState, Stack};
use nu_protocol::{
    Category, ListStream, PipelineData, ShellError, Signals, Signature, SyntaxShape, Type, Value,
};
use std::time::Duration;

use crate::nu::util;
use crate::store::{FollowOption, ReadOptions, Store};

#[derive(Clone)]
pub struct LastStreamCommand {
    store: Store,
}

impl LastStreamCommand {
    pub fn new(store: Store) -> Self {
        Self { store }
    }
}

impl Command for LastStreamCommand {
    fn name(&self) -> &str {
        ".last"
    }

    fn signature(&self) -> Signature {
        Signature::build(".last")
            .input_output_types(vec![(Type::Nothing, Type::Any)])
            .optional(
                "topic",
                SyntaxShape::String,
                "topic to get most recent frame from (default: all topics)",
            )
            .optional(
                "count",
                SyntaxShape::Int,
                "number of frames to return (default: 1)",
            )
            .switch(
                "follow",
                "long poll for updates to most recent frame",
                Some('f'),
            )
            .switch(
                "with-timestamp",
                "include timestamp extracted from frame ID",
                None,
            )
            .category(Category::Experimental)
    }

    fn description(&self) -> &str {
        "get the most recent frame(s) for a topic"
    }

    fn run(
        &self,
        engine_state: &EngineState,
        stack: &mut Stack,
        call: &Call,
        _input: PipelineData,
    ) -> Result<PipelineData, ShellError> {
        let raw_topic: Option<String> = call.opt(engine_state, stack, 0)?;
        let raw_count: Option<i64> = call.opt(engine_state, stack, 1)?;
        let follow = call.has_flag(engine_state, stack, "follow")?;
        let with_timestamp = call.has_flag(engine_state, stack, "with-timestamp")?;
        let span = call.head;

        // Disambiguate: if topic parses as a positive integer and count is absent,
        // treat it as the count (topics cannot start with digits per ADR 0002)
        let (topic, n) = match (&raw_topic, raw_count) {
            (Some(t), None) if t.parse::<usize>().is_ok() => (None, t.parse::<usize>().unwrap()),
            _ => (raw_topic, raw_count.map(|v| v as usize).unwrap_or(1)),
        };

        if !follow {
            // Non-follow mode: use sync path
            let options = ReadOptions::builder().last(n).maybe_topic(topic).build();

            let frames: Vec<Value> = self
                .store
                .read_sync(options)
                .map(|frame| util::frame_to_value(&frame, span, with_timestamp))
                .collect();

            return if frames.is_empty() {
                Ok(PipelineData::Empty)
            } else if frames.len() == 1 {
                Ok(PipelineData::Value(
                    frames.into_iter().next().unwrap(),
                    None,
                ))
            } else {
                Ok(PipelineData::Value(Value::list(frames, span), None))
            };
        }

        // Follow mode: use async path with streaming
        let options = ReadOptions::builder()
            .last(n)
            .maybe_topic(topic)
            .follow(FollowOption::On)
            .build();

        let store = self.store.clone();
        let signals = engine_state.signals().clone();

        // Create channel for async -> sync bridge
        let (tx, rx) = std::sync::mpsc::channel();

        // Spawn thread to handle async store.read()
        std::thread::spawn(move || {
            let rt = tokio::runtime::Runtime::new().expect("Failed to create tokio runtime");
            rt.block_on(async move {
                let mut receiver = store.read(options).await;

                while let Some(frame) = receiver.recv().await {
                    let value = util::frame_to_value(&frame, span, with_timestamp);

                    if tx.send(value).is_err() {
                        break;
                    }
                }
            });
        });

        // Create ListStream from channel with signal-aware polling
        let stream = ListStream::new(
            std::iter::from_fn(move || {
                use std::sync::mpsc::RecvTimeoutError;
                loop {
                    if signals.interrupted() {
                        return None;
                    }
                    match rx.recv_timeout(Duration::from_millis(100)) {
                        Ok(value) => return Some(value),
                        Err(RecvTimeoutError::Timeout) => continue,
                        Err(RecvTimeoutError::Disconnected) => return None,
                    }
                }
            }),
            span,
            Signals::empty(),
        );

        Ok(PipelineData::ListStream(stream, None))
    }
}