Brummel/RustAst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor: Replace Object with StreamStorage for streams

Browse Source 
The `Object` trait is too generic and has been causing confusion. This
commit introduces `StreamStorage` as a dedicated trait for reactive
stream types.

This change involves:

- Renaming `Object` to `StreamStorage` in relevant places.
- Updating the `Value::Object` enum variant to `Value::Stream`.
- Modifying how streams are handled in the compiler, VM, and tests to
  use the new `StreamStorage` trait.
- Adjusting example scripts and tests to reflect the change in type
  representation.
- The `StreamNode` struct now explicitly holds its `element_type`.
This commit is contained in:
Brummel
2026-03-29 18:01:46 +02:00
parent a665e2c1a5
commit 10a7fcb576
9 changed files with 235 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ast/closure.rs
+1 -15
View File
@@ -1,6 +1,5 @@
use crate::ast::nodes::{AnalyzedNode, ExecNode};
use crate::ast::types::{Object, Value};
use std::any::Any;
use crate::ast::types::Value;
use std::cell::RefCell;
use std::rc::Rc;
@@ -45,16 +44,3 @@ impl Closure {
}
}
impl Object for Closure {
fn type_name(&self) -> &'static str {
"closure"
}
fn as_any(&self) -> &dyn Any {
self
}
fn into_rc_any(self: Rc<Self>) -> Rc<dyn Any> {
self
}
}
src/ast/compiler/specializer.rs
+1 -1
View File
@@ -245,7 +245,7 @@ impl Specializer {
// Only replace the callee if the compiled value is actually a function/object.
// If it's a scalar (like 30 from folding), we DON'T fold here.
// We keep the Call but update the callee to the specialized version if it's an object.
if let Value::Object(_) | Value::Function(_) = &compiled_val {
if let Value::Stream(_) | Value::Function(_) = &compiled_val {
let specialized_callee = self.make_constant_node(
compiled_val,
StaticType::Function(Box::new(Signature {
src/ast/rtl/streams.rs
+143 -63
View File
@@ -1,7 +1,10 @@
use crate::ast::compiler::call_hooks::RtlCompilerHook;
use crate::ast::nodes::{Node, NodeKind, TypedNode, TypedPhase};
use crate::ast::rtl::series::{RingBuffer, ScalarValue, SeriesMember};
use crate::ast::types::{Object, PipeFn, Value};
use crate::ast::vm::VM;
use crate::ast::types::{NativeFunction, NodeIdentity, PipeFn, SourceLocation, StreamStorage, Value};
use crate::ast::vm::{GlobalStore, VM};
use std::cell::RefCell;
use std::collections::HashMap;
use std::rc::Rc;
/// A Signal is the "packet" flowing through the reactive pipeline.
@@ -52,10 +55,13 @@ impl<T: ObservableStream> ObservableStream for std::cell::RefCell<T> {
}
}
/// A generic wrapper to pass ANY ObservableStream (Root, Pipe, etc.) as an Object to the VM.
/// A generic wrapper to pass ANY ObservableStream (Root, Pipe, etc.) as a Stream to the VM.
#[derive(Clone)]
pub struct StreamNode {
pub inner: Rc<dyn ObservableStream>,
/// The `StaticType` of the elements emitted by this stream.
/// Set at construction time and used by `Value::static_type()`.
pub element_type: StaticType,
}
impl std::fmt::Debug for StreamNode {
@@ -64,9 +70,9 @@ impl std::fmt::Debug for StreamNode {
}
}
impl Object for StreamNode {
fn type_name(&self) -> &'static str {
"StreamNode"
impl StreamStorage for StreamNode {
fn stream_type_name(&self) -> &'static str {
"stream"
}
fn as_any(&self) -> &dyn std::any::Any {
self
@@ -74,6 +80,9 @@ impl Object for StreamNode {
fn into_rc_any(self: std::rc::Rc<Self>) -> std::rc::Rc<dyn std::any::Any> {
self
}
fn element_type(&self) -> StaticType {
self.element_type.clone()
}
}
/// The RootStream is the "Clock" and data source of the entire pipeline.
@@ -145,13 +154,23 @@ pub struct PipeStream {
pub executor: Option<Box<PipeFn>>,
/// Observers of THIS pipe.
observers: RefCell<Vec<Rc<RefCell<dyn Observer>>>>,
/// Output element type, injected by PipeHook::finalize. `Any` when unknown.
pub element_type: StaticType,
}
impl PipeStream {
pub fn new(
/// Creates a PipeStream with unknown output element type (`StaticType::Any`).
pub fn new(name: String, input_count: usize, executor: Option<Box<PipeFn>>) -> Self {
Self::new_typed(name, input_count, executor, StaticType::Any)
}
/// Creates a PipeStream with a known output element type.
/// Called by `build_pipeline_node` when the type is available from `PipeHook::finalize`.
pub fn new_typed(
name: String,
input_count: usize,
executor: Option<Box<PipeFn>>,
element_type: StaticType,
) -> Self {
Self {
name,
@@ -161,6 +180,7 @@ impl PipeStream {
current_signal: RefCell::new(None),
executor,
observers: RefCell::new(Vec::new()),
element_type,
}
}
}
@@ -268,12 +288,13 @@ impl Observer for RecordPusher {
pub fn build_pipeline_node(
inputs: Vec<Rc<dyn ObservableStream>>,
executor: Box<PipeFn>,
_out_type: &StaticType,
out_type: &StaticType,
) -> Rc<StreamNode> {
let pipe = Rc::new(RefCell::new(PipeStream::new(
let pipe = Rc::new(RefCell::new(PipeStream::new_typed(
"pipe".to_string(),
inputs.len(),
Some(executor),
out_type.clone(),
)));
// Connect inputs to the pipe
@@ -285,7 +306,7 @@ pub fn build_pipeline_node(
input.add_observer(adapter);
}
Rc::new(StreamNode { inner: pipe })
Rc::new(StreamNode { inner: pipe, element_type: out_type.clone() })
}
pub fn build_map_stream(input: Rc<dyn ObservableStream>, field: Keyword) -> StreamNode {
@@ -313,6 +334,7 @@ pub fn build_map_stream(input: Rc<dyn ObservableStream>, field: Keyword) -> Stre
StreamNode {
inner: pipe,
element_type: StaticType::Any,
}
}
@@ -323,6 +345,98 @@ pub fn build_map_stream(input: Rc<dyn ObservableStream>, field: Keyword) -> Stre
use crate::ast::environment::Environment;
use crate::ast::types::{Keyword, Purity, RecordLayout, Signature, StaticType};
/// Extracts `ObservableStream` references from a runtime `Value`.
/// Accepts a single `StreamNode` or a `Tuple` of `StreamNode`s.
fn extract_obs_streams(val: &Value) -> Vec<Rc<dyn ObservableStream>> {
match val {
Value::Tuple(elements) => elements
.iter()
.map(|v| {
if let Value::Stream(s) = v
&& let Some(sn) = s.as_any().downcast_ref::<StreamNode>()
{
sn.inner.clone()
} else {
panic!("pipe: each input must be a StreamNode");
}
})
.collect(),
Value::Stream(s) => {
if let Some(sn) = s.as_any().downcast_ref::<StreamNode>() {
vec![sn.inner.clone()]
} else {
panic!("pipe: input must be a StreamNode");
}
}
_ => panic!("pipe: first argument must be a stream or tuple of streams"),
}
}
/// Builds a `PipeFn` executor from a runtime `Value::Closure` or `Value::Function`.
fn build_pipe_executor(val: &Value, globals: GlobalStore) -> Box<PipeFn> {
match val {
Value::Closure(rc) => {
let my_closure = rc.clone();
let mut pipe_vm = VM::new(globals);
Box::new(move |call_args: &[Value]| -> Value {
pipe_vm
.run_with_args(my_closure.clone(), call_args)
.unwrap_or_else(|e| panic!("Pipeline lambda execution failed: {}", e))
})
}
Value::Function(f) => {
let my_func = f.clone();
Box::new(move |call_args: &[Value]| -> Value { (my_func.func)(call_args) })
}
_ => panic!("pipe: second argument must be a function or closure"),
}
}
/// Compiler hook for `(pipe inputs lambda)`.
///
/// - **finalize:** Replaces the `pipe` identifier with a pre-configured factory closure
/// that captures both the resolved output element type and the global store. This mirrors
/// `SeriesHook::finalize` and makes the element type available to `build_pipeline_node`
/// at runtime without a runtime type lookup.
pub struct PipeHook {
globals: GlobalStore,
}
impl RtlCompilerHook for PipeHook {
fn finalize(
&self,
_callee: Rc<TypedNode>,
args: Rc<TypedNode>,
node_ty: &StaticType,
_subst: &HashMap<u32, StaticType>,
) -> Option<NodeKind<TypedPhase>> {
let StaticType::Stream(inner) = node_ty else { return None };
// Only finalize when the element type is concrete — not unresolved Any or TypeVar.
if matches!(inner.as_ref(), StaticType::Any | StaticType::TypeVar(_)) {
return None;
}
let element_type = *inner.clone();
let globals = self.globals.clone();
let factory: Value = Value::Function(Rc::new(NativeFunction {
func: Rc::new(move |call_args: &[Value]| {
let obs = extract_obs_streams(&call_args[0]);
let exec = build_pipe_executor(&call_args[1], globals.clone());
Value::Stream(build_pipeline_node(obs, exec, &element_type))
}),
purity: Purity::Impure,
}));
let factory_node = Rc::new(Node {
kind: NodeKind::Constant(factory),
ty: StaticType::Any,
identity: NodeIdentity::new(SourceLocation { line: 0, col: 0 }),
comments: Rc::from([]),
});
Some(NodeKind::Call { callee: factory_node, args })
}
}
/// Resolves the return type of a `pipe` call from its argument types.
/// Argument layout: `Tuple([inputs, lambda])` where inputs is a Tuple of streams
/// or a single StreamNode, and lambda is a `Function<args -> U>`.
@@ -412,11 +526,8 @@ pub fn register(env: &Environment) {
// 1. Create the RootStream
let root_stream = Rc::new(RootStream::new());
let stream_node = StreamNode {
inner: root_stream.clone(),
};
// 2. Setup the Layout for OHLC records
// 2. Setup the Layout for OHLC records (before StreamNode so element_type is available)
let layout = RecordLayout::get_or_create(vec![
(Keyword::intern("open"), StaticType::Float),
(Keyword::intern("high"), StaticType::Float),
@@ -424,6 +535,11 @@ pub fn register(env: &Environment) {
(Keyword::intern("close"), StaticType::Float),
]);
let stream_node = StreamNode {
inner: root_stream.clone(),
element_type: StaticType::Record(layout.clone()),
};
// 3. Create the stateful generator closure
let mut current_tick = 0;
let mut last_close = 100.0;
@@ -465,7 +581,7 @@ pub fn register(env: &Environment) {
generators.borrow_mut().push(Box::new(generator));
// 5. Return the stream reference to the script
Value::Object(Rc::new(stream_node))
Value::Stream(Rc::new(stream_node))
},
).doc("Creates a stream of random OHLC (Open-High-Low-Close) candles.")
.description("Args: (seed: int, limit: int). Uses a random walk model. Connect via (pipe ...).")
@@ -500,6 +616,7 @@ pub fn register(env: &Environment) {
let root_stream = Rc::new(RootStream::new());
let stream_node = StreamNode {
inner: root_stream.clone(),
element_type: StaticType::Bool,
};
// 2. Setup isolated VM
@@ -528,7 +645,7 @@ pub fn register(env: &Environment) {
ticker_generators.borrow_mut().push(Box::new(generator));
// 5. Return stream
Value::Object(Rc::new(stream_node))
Value::Stream(Rc::new(stream_node))
},
).doc("Creates a stream driven by a boolean closure. Ticks as long as closure returns true.")
.examples(&["(def ticker (create-ticker (fn [] (< (now) end-time))))"]);
@@ -537,7 +654,9 @@ pub fn register(env: &Environment) {
// inputs: a Tuple of StreamNodes or a single StreamNode
// lambda: a Closure or Function
// Same RTL-only bootstrap capture as create-ticker above.
let globals = env.global_store();
// `fn_globals` is moved into the native fn fallback; `hook_globals` is captured by PipeHook.
let fn_globals = env.global_store();
let hook_globals = fn_globals.clone();
env.register_native_fn(
"pipe",
StaticType::PolymorphicFn {
@@ -546,54 +665,15 @@ pub fn register(env: &Environment) {
},
Purity::Impure,
move |args: &[Value]| {
// This fallback only runs when PipeHook::finalize did not replace the callee
// (e.g. the output element type could not be determined at compile time).
assert!(args.len() == 2, "pipe expects exactly 2 arguments (inputs, lambda)");
// Extract observable streams from args[0]
let obs_streams: Vec<Rc<dyn ObservableStream>> = match &args[0] {
Value::Tuple(elements) => elements
.iter()
.map(|v| {
if let Value::Object(obj) = v
&& let Some(sn) = obj.as_any().downcast_ref::<StreamNode>()
{
sn.inner.clone()
} else {
panic!("pipe: each input must be a StreamNode");
}
})
.collect(),
Value::Object(obj) => {
if let Some(sn) = obj.as_any().downcast_ref::<StreamNode>() {
vec![sn.inner.clone()]
} else {
panic!("pipe: input must be a StreamNode");
}
}
_ => panic!("pipe: first argument must be a stream or tuple of streams"),
};
// Build the executor closure from args[1]
let mut pipe_vm = VM::new(globals.clone());
let executor: Box<PipeFn> = match &args[1] {
Value::Closure(rc) => {
let my_closure = rc.clone();
Box::new(move |call_args: &[Value]| -> Value {
pipe_vm
.run_with_args(my_closure.clone(), call_args)
.unwrap_or_else(|e| panic!("Pipeline lambda execution failed: {}", e))
})
}
Value::Function(f) => {
let my_func = f.clone();
Box::new(move |call_args: &[Value]| -> Value { (my_func.func)(call_args) })
}
_ => panic!("pipe: second argument must be a function or closure"),
};
let node = build_pipeline_node(obs_streams, executor, &StaticType::Any);
Value::Object(node)
let obs = extract_obs_streams(&args[0]);
let exec = build_pipe_executor(&args[1], fn_globals.clone());
Value::Stream(build_pipeline_node(obs, exec, &StaticType::Any))
},
).doc("Connects a lambda to one or more streams, producing a transformed output stream.")
).with_compiler_hook(Rc::new(PipeHook { globals: hook_globals }))
.doc("Connects a lambda to one or more streams, producing a transformed output stream.")
.description("Returns void from lambda to act as a filter (value is dropped). Supports tuple inputs for barrier synchronization.")
.examples(&[
"(pipe ohlc (fn [bar] (.close bar)))",
src/ast/types.rs
+16 -15
View File
@@ -192,17 +192,6 @@ impl RecordLayout {
}
}
/// Interface for custom RTL objects (Streams and custom extensions).
/// For series data, use `Value::Series` with the `SeriesStorage` trait instead.
pub trait Object: fmt::Debug {
fn type_name(&self) -> &'static str;
fn as_any(&self) -> &dyn Any;
/// Converts `Rc<Self>` into `Rc<dyn Any>`, enabling zero-copy downcasting to a concrete
/// `Rc<T>` via `Rc::downcast::<T>()`. Requires `arbitrary_self_types` (stable since 1.81).
fn into_rc_any(self: Rc<Self>) -> Rc<dyn Any>;
}
/// Read interface for all series types: indexed lookback access and type metadata.
pub trait SeriesStorage: fmt::Debug {
fn series_type_name(&self) -> &'static str;
@@ -241,6 +230,18 @@ pub trait PushableStorage: SeriesStorage {
fn push_value(&self, val: Value);
}
/// Read interface for reactive stream objects (StreamNode and derived streams).
pub trait StreamStorage: fmt::Debug {
fn stream_type_name(&self) -> &'static str;
fn as_any(&self) -> &dyn Any;
/// Converts `Rc<Self>` into `Rc<dyn Any>` for zero-copy concrete downcast.
fn into_rc_any(self: Rc<Self>) -> Rc<dyn Any>;
/// Returns the `StaticType` of the elements emitted by this stream.
fn element_type(&self) -> StaticType;
}
/// A shared sequence of values, used by both Tuples and Records.
pub type ValueList = Rc<Vec<Value>>;
@@ -287,7 +288,7 @@ pub enum Value {
Closure(Rc<Closure>), // Compiled function with captured upvalues
Quote(Rc<SyntaxNode>), // Quoted AST node (from 'expr or macro expansion)
Series(Rc<dyn SeriesStorage>), // Time series: ScalarSeries, RecordSeries, SeriesView, etc.
Object(Rc<dyn Object>), // RTL extensions: Streams and custom types
Stream(Rc<dyn StreamStorage>), // Reactive stream: StreamNode and derived streams
Cell(Rc<RefCell<Value>>), // Boxed value for captures
}
@@ -310,7 +311,7 @@ impl PartialEq for Value {
(Value::Closure(a), Value::Closure(b)) => Rc::ptr_eq(a, b),
(Value::Quote(a), Value::Quote(b)) => Rc::ptr_eq(a, b),
(Value::Series(a), Value::Series(b)) => Rc::ptr_eq(a, b),
(Value::Object(a), Value::Object(b)) => Rc::ptr_eq(a, b),
(Value::Stream(a), Value::Stream(b)) => Rc::ptr_eq(a, b),
(Value::Cell(a), Value::Cell(b)) => Rc::ptr_eq(a, b),
_ => false,
}
@@ -747,7 +748,7 @@ impl Value {
Value::Closure(_) => StaticType::Object("closure"),
Value::Quote(_) => StaticType::Object("ast-node"),
Value::Series(s) => StaticType::Series(Box::new(s.inner_static_type())),
Value::Object(o) => StaticType::Object(o.type_name()),
Value::Stream(s) => StaticType::Stream(Box::new(s.element_type())),
Value::Cell(c) => c.borrow().static_type(),
}
}
@@ -793,7 +794,7 @@ impl fmt::Display for Value {
Value::Closure(_) => write!(f, "<closure>"),
Value::Quote(_) => write!(f, "<ast-node>"),
Value::Series(s) => write!(f, "<series:{}>", s.series_type_name()),
Value::Object(o) => write!(f, "<{:?}>", o),
Value::Stream(s) => write!(f, "<stream:{}>", s.stream_type_name()),
Value::Cell(c) => write!(f, "{}", c.borrow()),
}
}
src/ast/vm.rs
+18 -24
View File
@@ -209,11 +209,8 @@ impl VM {
));
}
}
Value::Object(obj) => {
return Err(format!(
"Tail call target is not callable: {}",
obj.type_name()
));
Value::Stream(_) => {
return Err("Tail call target is not callable: stream".to_string());
}
other => {
return Err(format!("Tail call target is not callable: {}", other));
@@ -411,15 +408,12 @@ impl VM {
s.series_type_name()
))
}
Value::Object(obj) => {
if let Some(sn) = obj.as_any().downcast_ref::<StreamNode>() {
Value::Stream(s) => {
if let Some(sn) = s.as_any().downcast_ref::<StreamNode>() {
let mapped = build_map_stream(sn.inner.clone(), *field);
return Ok(Value::Object(Rc::new(mapped)));
return Ok(Value::Stream(Rc::new(mapped)));
}
Err(format!(
"Attempt to access field on non-record object: {}",
obj.type_name()
))
Err("Attempt to access field on non-record stream".to_string())
}
_ => Err(format!(
"Attempt to access field on non-record: {}",
@@ -483,7 +477,7 @@ impl VM {
self.stack.truncate(base);
match func_val {
Value::Closure(_) | Value::Object(_) | Value::Series(_) => {
Value::Closure(_) | Value::Stream(_) | Value::Series(_) => {
self.tail_call = Some((func_val, arg_vals));
return Ok(Value::Void);
}
@@ -535,15 +529,15 @@ impl VM {
k.name(),
s.series_type_name()
));
} else if let Value::Object(obj) = rec {
if let Some(sn) = obj.as_any().downcast_ref::<StreamNode>() {
} else if let Value::Stream(s) = rec {
if let Some(sn) = s.as_any().downcast_ref::<StreamNode>() {
let mapped = build_map_stream(sn.inner.clone(), k);
return Ok(Value::Object(Rc::new(mapped)));
return Ok(Value::Stream(Rc::new(mapped)));
}
return Err(format!(
"Field accessor .{} expects a record, RecordSeries or Stream, got {}",
k.name(),
obj.type_name()
s.stream_type_name()
));
} else {
return Err(format!(
@@ -616,15 +610,15 @@ impl VM {
s.series_type_name()
))
}
} else if let Value::Object(obj) = rec {
if let Some(sn) = obj.as_any().downcast_ref::<StreamNode>() {
} else if let Value::Stream(s) = rec {
if let Some(sn) = s.as_any().downcast_ref::<StreamNode>() {
let mapped = build_map_stream(sn.inner.clone(), *k);
Ok(Value::Object(Rc::new(mapped)))
Ok(Value::Stream(Rc::new(mapped)))
} else {
Err(format!(
"Field accessor .{} expects a record, RecordSeries or Stream, got {}",
k.name(),
obj.type_name()
s.stream_type_name()
))
}
} else {
@@ -713,9 +707,9 @@ impl VM {
self.stack.truncate(base);
return res;
}
Value::Object(obj) => {
Value::Stream(_) => {
self.stack.truncate(base);
return Err(format!("Object is not callable: {}", obj.type_name()));
return Err("Stream is not callable".to_string());
}
_ => {
self.stack.truncate(base);
@@ -734,7 +728,7 @@ impl VM {
Value::Closure(_) => "Closure",
Value::Quote(_) => "Quote",
Value::Series(_) => "Series",
Value::Object(_) => "Object",
Value::Stream(_) => "Stream",
Value::Cell(_) => "Cell",
};
return Err(format!("Attempt to call non-function: {} (Variant: {})", current_func, variant_name));