diff --git a/src/ast/rtl/series.rs b/src/ast/rtl/series.rs index cc86741..65b0011 100644 --- a/src/ast/rtl/series.rs +++ b/src/ast/rtl/series.rs @@ -386,6 +386,36 @@ impl crate::ast::types::Series for SharedSeries { } } +/// A read-only series for generic Values that shares its buffer with the pipeline. +#[derive(Clone)] +pub struct SharedValueSeries { + pub buffer: Rc>>, +} + +impl Debug for SharedValueSeries { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "SharedValueSeries[len: {}]", self.buffer.borrow().len()) + } +} + +impl Object for SharedValueSeries { + fn type_name(&self) -> &'static str { + "SharedValueSeries" + } + fn as_any(&self) -> &dyn Any { + self + } + fn as_series(&self) -> Option<&dyn crate::ast::types::Series> { + Some(self) + } +} + +impl crate::ast::types::Series for SharedValueSeries { + fn get_item(&self, index: usize) -> Option { + self.buffer.borrow().get(index).cloned() + } +} + /// A read-only series for Records that shares its buffers with the pipeline. #[derive(Clone)] pub struct SharedRecordSeries { diff --git a/src/ast/rtl/streams.rs b/src/ast/rtl/streams.rs index d0bb831..a6bc2ad 100644 --- a/src/ast/rtl/streams.rs +++ b/src/ast/rtl/streams.rs @@ -29,6 +29,12 @@ pub trait ObservableStream { fn add_observer(&self, observer: Rc>); } +impl ObservableStream for std::cell::RefCell { + fn add_observer(&self, observer: Rc>) { + self.borrow().add_observer(observer); + } +} + /// A generic wrapper to pass ANY ObservableStream (Root, Pipe, etc.) as an Object to the VM. #[derive(Clone)] pub struct StreamNode { @@ -50,6 +56,32 @@ impl crate::ast::types::Object for StreamNode { } } +/// A node that represents the result of a `pipe` statement. +/// It acts as a Stream (to connect downstream pipes) AND as a Series (for script lookbacks). +#[derive(Clone)] +pub struct PipelineNode { + pub stream: Rc, + pub series: Rc, +} + +impl std::fmt::Debug for PipelineNode { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "PipelineNode") + } +} + +impl crate::ast::types::Object for PipelineNode { + fn type_name(&self) -> &'static str { + "PipelineNode" + } + fn as_any(&self) -> &dyn std::any::Any { + self + } + fn as_series(&self) -> Option<&dyn crate::ast::types::Series> { + Some(self.series.as_ref()) + } +} + /// The RootStream is the "Clock" and data source of the entire pipeline. /// It generates the monotonic `cycle_id` and triggers the observers. pub struct RootStream { @@ -160,13 +192,20 @@ impl Observer for PipeStream { let args = vec![value]; // Simplified for now // 2. Execute Lambda (This requires a VM instance!) + // TODO: Actually execute the lambda using a VM context. + // For now, we mock the result to be the input value itself. let result = args[0].clone(); - // 3. Update Current Signal + // 3. Handle Void case! (Filter pattern) + if matches!(result, Value::Void) { + return; // Act as a filter: do not emit, do not push. + } + + // 4. Update Current Signal let new_signal = Signal { cycle_id, value: result }; *self.current_signal.borrow_mut() = Some(new_signal.clone()); - // 4. Notify Observers (Always at source_index 0 of the NEXT pipe) + // 5. Notify Observers (Always at source_index 0 of the NEXT pipe) let obs_list = self.observers.borrow(); for obs in obs_list.iter() { obs.borrow_mut().notify(0, cycle_id, new_signal.value.clone()); @@ -187,6 +226,18 @@ impl Observer for SeriesPusher { } } +/// A specialized observer that pushes incoming signals into a generic SharedValueSeries buffer. +pub struct ValuePusher { + pub buffer: Rc>>, + pub lookback: Option, +} + +impl Observer for ValuePusher { + fn notify(&mut self, _source_index: usize, _cycle_id: u64, value: Value) { + self.buffer.borrow_mut().push(value, self.lookback); + } +} + // ============================================================================ // Script Integration (RTL Registration) // ============================================================================ diff --git a/src/ast/vm.rs b/src/ast/vm.rs index 591ee91..244fcca 100644 --- a/src/ast/vm.rs +++ b/src/ast/vm.rs @@ -369,15 +369,63 @@ impl VM { } } BoundKind::Pipe { inputs, lambda } => { - // To be implemented in next step: - // 1. Create PipeStream - // 2. Link with RootStream and Inputs - // 3. Create and return SharedSeries + use crate::ast::rtl::streams::{PipelineNode, PipeStream, StreamNode, ValuePusher, ObservableStream}; + use crate::ast::rtl::series::{RingBuffer, SharedValueSeries}; + + let mut obs_streams = Vec::new(); for input in inputs { - self.eval_internal(obs, input)?; + let val = self.eval_internal(obs, input)?; + if let Value::Object(obj) = val { + if let Some(s) = obj.as_any().downcast_ref::() { + obs_streams.push(s.inner.clone()); + } else if let Some(p) = obj.as_any().downcast_ref::() { + obs_streams.push(p.stream.clone()); + } else { + return Err(format!("Pipe input must be a stream, found {}", obj.type_name())); + } + } else { + return Err("Pipe input must be an object (stream)".to_string()); + } } - self.eval_internal(obs, lambda)?; - Ok(Value::Void) + + let lambda_val = self.eval_internal(obs, lambda)?; + let lambda_obj = if let Value::Object(obj) = lambda_val { + obj + } else { + return Err("Pipe lambda must be a function/closure".to_string()); + }; + + let pipe = Rc::new(RefCell::new(PipeStream::new( + "pipe".to_string(), + inputs.len(), + Some(lambda_obj) + ))); + + for stream in obs_streams { + stream.add_observer(pipe.clone() as Rc>); + } + + // Create the data buffer and pusher for the pipe's output + let buffer = Rc::new(RefCell::new(RingBuffer::::new())); + let pusher = Rc::new(RefCell::new(ValuePusher { + buffer: buffer.clone(), + lookback: Some(100), // Default lookback for now + })); + + // The pipe pushes to the buffer + pipe.borrow().add_observer(pusher); + + // Create the Series view for script access + let series = Rc::new(SharedValueSeries { + buffer, + }); + + let node = Rc::new(PipelineNode { + stream: pipe, + series, + }); + + Ok(Value::Object(node)) } BoundKind::Block { exprs } => { let mut last = Value::Void;