From 3676eb51e54e23318bb7dea51c1270ece328487c Mon Sep 17 00:00:00 2001 From: Brummel Date: Mon, 30 Mar 2026 14:16:04 +0200 Subject: [PATCH] Rename pipe-buffered to pipe-series The `pipe-buffered` function has been renamed to `pipe-series` to better reflect its functionality. This change involves updating the function name in the example file `examples/data_stream_pipe_buffered.myc` and across the Rust source code in `src/ast/rtl/streams.rs`. Additionally, test cases and documentation have been updated accordingly. --- ...ffered.myc => data_stream_pipe_series.myc} | 2 +- src/ast/rtl/streams.rs | 28 +++++++++---------- tests/pipeline.rs | 24 ++++++++-------- 3 files changed, 27 insertions(+), 27 deletions(-) rename examples/{data_stream_pipe_buffered.myc => data_stream_pipe_series.myc} (93%) diff --git a/examples/data_stream_pipe_buffered.myc b/examples/data_stream_pipe_series.myc similarity index 93% rename from examples/data_stream_pipe_buffered.myc rename to examples/data_stream_pipe_series.myc index d89656f..f07cc2b 100644 --- a/examples/data_stream_pipe_buffered.myc +++ b/examples/data_stream_pipe_series.myc @@ -5,7 +5,7 @@ (def cnt 1) - (pipe-buffered 2 [EURUSD GER40] + (pipe-series 2 [EURUSD GER40] (fn [eu ge] (do (def dax (.close (ge 0))) diff --git a/src/ast/rtl/streams.rs b/src/ast/rtl/streams.rs index 618a313..f4edbc5 100644 --- a/src/ast/rtl/streams.rs +++ b/src/ast/rtl/streams.rs @@ -529,10 +529,10 @@ fn pipe_arg_hint_resolver( } // ============================================================================ -// pipe-buffered: Pipe with automatic value accumulation into Series +// pipe-series: Pipe with automatic value accumulation into Series // ============================================================================ -/// Compiler hook for `(pipe-buffered lookback inputs lambda)`. +/// Compiler hook for `(pipe-series lookback inputs lambda)`. /// /// - **post_call:** Validates 3 arguments (Int, Streams, Lambda) and checks that /// the input stream count matches the lambda parameter count. @@ -551,7 +551,7 @@ impl RtlCompilerHook for LookbackPipeHook { _ctx: &dyn InferenceAccess, diag: &mut Diagnostics, ) -> StaticType { - // Validate: (pipe-buffered Int Streams Lambda) — 3 elements + // Validate: (pipe-series Int Streams Lambda) — 3 elements let NodeKind::Tuple { elements } = &args.kind else { return ret_ty }; if elements.len() != 3 || !matches!(elements[0].ty, StaticType::Int) { return ret_ty; @@ -572,7 +572,7 @@ impl RtlCompilerHook for LookbackPipeHook { if actual != expected { diag.push_error( format!( - "pipe-buffered: lambda expects {} parameter(s) but {} input stream(s) provided", + "pipe-series: lambda expects {} parameter(s) but {} input stream(s) provided", actual, expected ), Some(elements[2].identity.clone()), @@ -655,7 +655,7 @@ fn build_buffered_wrapper( }) } -/// Extracts input element types from the typed AST for `pipe-buffered`. +/// Extracts input element types from the typed AST for `pipe-series`. /// Args layout: `Tuple([Int, inputs, Lambda])` — inputs at index 1. fn extract_input_element_types(args: &TypedNode) -> Vec { let NodeKind::Tuple { elements } = &args.kind else { return vec![] }; @@ -694,7 +694,7 @@ fn extract_runtime_input_types(val: &Value) -> Vec { } } -/// Resolves the return type of `pipe-buffered` from its argument types. +/// Resolves the return type of `pipe-series` from its argument types. /// Argument layout: `Tuple([Int, inputs, Lambda])`. /// Returns `Stream`, unwrapping `Optional` to `U` (filter pattern). fn buffered_pipe_type_resolver(args_ty: &StaticType) -> Option { @@ -712,7 +712,7 @@ fn buffered_pipe_type_resolver(args_ty: &StaticType) -> Option { Some(StaticType::Stream(Box::new(inner))) } -/// Provides expected lambda parameter types for `pipe-buffered`. +/// Provides expected lambda parameter types for `pipe-series`. /// Lambda is at arg index 2. Parameters are wrapped as `Series` instead of raw `T`. fn buffered_pipe_arg_hint_resolver( arg_index: usize, @@ -939,21 +939,21 @@ pub fn register(env: &Environment) { "(pipe [stream-a stream-b] (fn [a b] (+ a b)))", ]); - // (pipe-buffered lookback inputs lambda) -> StreamNode + // (pipe-series lookback inputs lambda) -> StreamNode // Like pipe, but accumulates values into internal Series with the given lookback // depth. The lambda receives Series objects and only fires once all series have // at least `lookback` elements (fill gate). let fn_globals = env.global_store(); let hook_globals = fn_globals.clone(); env.register_native_fn( - "pipe-buffered", + "pipe-series", StaticType::PolymorphicFn { resolve_return: buffered_pipe_type_resolver, resolve_arg_hints: Some(buffered_pipe_arg_hint_resolver), }, Purity::Impure, move |args: &[Value]| { - assert!(args.len() == 3, "pipe-buffered expects 3 arguments (lookback, inputs, lambda)"); + assert!(args.len() == 3, "pipe-series expects 3 arguments (lookback, inputs, lambda)"); let lookback = args[0].as_int().unwrap() as usize; let obs = extract_obs_streams(&args[1]); let input_types: Vec = extract_runtime_input_types(&args[1]); @@ -965,8 +965,8 @@ pub fn register(env: &Environment) { .doc("Like pipe, but accumulates values into Series before firing the lambda.") .description("The lambda only fires once all input series have at least N elements (fill gate). Lambda parameters are Series objects with lookback indexing (0 = newest).") .examples(&[ - "(pipe-buffered 20 ohlc (fn [bars] (- (.close (bars 0)) (.close (bars 19)))))", - "(pipe-buffered 2 [stream-a stream-b] (fn [a b] (+ (a 0) (b 0))))", + "(pipe-series 20 ohlc (fn [bars] (- (.close (bars 0)) (.close (bars 19)))))", + "(pipe-series 2 [stream-a stream-b] (fn [a b] (+ (a 0) (b 0))))", ]); } @@ -1034,7 +1034,7 @@ mod tests { assert_eq!(sig.cycle_id, 1); } - /// Validates the wrapper-executor pattern for `pipe-buffered`: + /// Validates the wrapper-executor pattern for `pipe-series`: /// A closure wraps push + fill gate + user lambda, reusing standard PipeStream. #[test] fn test_buffered_pipe_fill_gate() { @@ -1043,7 +1043,7 @@ mod tests { let root = RootStream::new(); - // Internal series with lookback 3 (simulates what pipe-buffered creates) + // Internal series with lookback 3 (simulates what pipe-series creates) let series: Rc = Rc::new(ScalarSeries::::new("FloatSeries", 3)); let series_clone = Rc::clone(&series); diff --git a/tests/pipeline.rs b/tests/pipeline.rs index 85a1984..d3ceefc 100644 --- a/tests/pipeline.rs +++ b/tests/pipeline.rs @@ -146,57 +146,57 @@ fn test_pipeline_optional_type() { } } -/// Verifies that `pipe-buffered` accumulates values into Series and only fires +/// Verifies that `pipe-series` accumulates values into Series and only fires /// the lambda once all series have reached the lookback depth (fill gate). /// Lambda parameters must be Series objects, not individual values. #[test] -fn test_pipe_buffered_fill_gate() { +fn test_pipe_series_fill_gate() { let env = Environment::new(); // create-random-ohlc produces 10 ticks. With lookback 3, the lambda // fires from tick 3 onwards (8 times). Each call returns the difference // between the current and previous close: (.close (bars 0)) - (.close (bars 1)) let source = "(do (def ohlc (create-random-ohlc 42 10)) - (def result (pipe-buffered 3 [ohlc] + (def result (pipe-series 3 [ohlc] (fn [bars] (- (.close (bars 0)) (.close (bars 1)))))) result )"; let res = env.run_script(source); - assert!(res.is_ok(), "pipe-buffered script failed: {:?}", res.err()); + assert!(res.is_ok(), "pipe-series script failed: {:?}", res.err()); let val = res.unwrap(); if let Value::Stream(s) = val { assert_eq!(s.stream_type_name(), "stream"); } else { - panic!("Expected Stream from pipe-buffered, got: {:?}", val); + panic!("Expected Stream from pipe-series, got: {:?}", val); } } -/// Verifies that pipe-buffered with multiple inputs waits for all series to fill. +/// Verifies that pipe-series with multiple inputs waits for all series to fill. #[test] -fn test_pipe_buffered_multi_input() { +fn test_pipe_series_multi_input() { let env = Environment::new(); let source = "(do (def src1 (create-random-ohlc 42 10)) (def src2 (create-random-ohlc 99 10)) - (def result (pipe-buffered 2 [src1 src2] + (def result (pipe-series 2 [src1 src2] (fn [a b] (+ (.close (a 0)) (.close (b 0)))))) result )"; let res = env.run_script(source); - assert!(res.is_ok(), "pipe-buffered multi-input failed: {:?}", res.err()); + assert!(res.is_ok(), "pipe-series multi-input failed: {:?}", res.err()); } -/// Verifies that pipe-buffered lambda params are typed as Series, not raw values. +/// Verifies that pipe-series lambda params are typed as Series, not raw values. /// The arg hint resolver must wrap inner types in Series. #[test] -fn test_pipe_buffered_lambda_params_are_series() { +fn test_pipe_series_lambda_params_are_series() { let env = Environment::new(); let source = "(do (def ohlc (create-random-ohlc 42 5)) - (pipe-buffered 3 [ohlc] (fn [bars] (.close (bars 0)))) + (pipe-series 3 [ohlc] (fn [bars] (.close (bars 0)))) )"; let dump = env.dump_ast(source).expect("dump_ast failed");