Struct ArrayHandle 

pub struct ArrayHandle<'a, T = StoreEntry> { /* private fields */ }

A draft-time builder for populating an array entry in the Store.

Returned by Pipeline::array and Store::define_array. Methods are chainable so nested structures can be built in a single expression. Nested arrays and maps are added via push_array and push_map and return child handles borrowing from this one.

Implementations

impl<'a> ArrayHandle<'a, StoreEntry>

pub fn push_array(&mut self) -> Result<ArrayHandle<'_, StoreEntry>, StoreError>

Appends a new nested array entry and returns a child handle borrowing from this one to populate it.

pub fn push_map(&mut self) -> Result<MapHandle<'_, StoreEntry>, StoreError>

Appends a new nested map entry and returns a child handle borrowing from this one to populate it.

Examples found in repository

examples/extend_store_nested.rs (line 13)

fn write_query_results(store: &mut Store) -> Result<(), Box<dyn std::error::Error>> {
    store.with_array("rows", |rows| {
        // Row 1 — all fields inserted via chaining, no .into() needed.
        let mut row = rows.push_map()?;
        row.insert("user", "alice@contoso.com")?
            .insert("ip", "10.0.0.1")?
            .insert("status", "Success")?
            .insert("attempts", 1i64)?;

        // Row 2
        let mut row = rows.push_map()?;
        row.insert("user", "bob@contoso.com")?
            .insert("ip", "192.168.1.50")?
            .insert("status", "Failure")?
            .insert("attempts", 3i64)?;

        Ok(())
    })?;

    Ok(())
}

fn write_grouped_data(store: &mut Store) -> Result<(), Box<dyn std::error::Error>> {
    store.with_map("by_severity", |groups| {
        groups.with_array("high", |arr| {
            arr.push("INC-001")?.push("INC-004")?;
            Ok(())
        })?;

        groups.with_array("medium", |arr| {
            arr.push("INC-002")?;
            Ok(())
        })?;

        groups.with_array("low", |arr| {
            arr.push("INC-003")?.push("INC-005")?.push("INC-006")?;
            Ok(())
        })?;

        Ok(())
    })?;

    Ok(())
}

fn write_nested_config(store: &mut Store) -> Result<(), Box<dyn std::error::Error>> {
    store.with_map("pipeline_config", |cfg| {
        cfg.insert("name", "signin_analysis")?;

        cfg.with_map("source", |src| {
            src.insert("type", "kql")?
                .insert("workspace", "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx")?;

            src.with_map("options", |opts| {
                opts.insert("timeout_secs", 30i64)?
                    .insert("max_rows", 10_000i64)?;
                Ok(())
            })?;

            Ok(())
        })?;

        cfg.with_array("outputs", |outputs| {
            let mut o = outputs.push_map()?;
            o.insert("format", "json")?
                .insert("path", "/tmp/results.json")?;

            let mut o = outputs.push_map()?;
            o.insert("format", "csv")?
                .insert("path", "/tmp/results.csv")?;

            Ok(())
        })?;

        Ok(())
    })?;

    Ok(())
}

examples/extend_store_collections.rs (line 80)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut store = Store::new();

    // --- Chainable push ---
    // push() accepts V: Into<StoreEntry>. The blanket impl
    // `From<T: Into<Value>> for StoreEntry` means primitives work directly.
    let mut arr = store.define_array("numbers")?.push(1)?.push(2)?.push(3)?;

    let items = store.get("numbers")?.as_array()?;
    println!("numbers: len={}", items.len());
    assert_eq!(items.len(), 3);

    // --- Chainable insert ---
    // insert() accepts K: Into<String> and V: Into<StoreEntry>.
    let mut map = store.define_map("config")?;
    map.insert("host", "localhost")?
        .insert("port", 8080i64)?
        .insert("debug", true)?;

    let host = store.get("config")?.get_key("host")?.get_value()?;
    println!("config.host = {host:?}");

    // --- Closure API on Store ---
    // with_array scopes the handle so you don't need an explicit drop.
    store.with_array("tags", |arr| {
        arr.push("rust")?.push("pipeline")?.push("store")?;
        Ok(())
    })?;

    let tags = store.get("tags")?.as_array()?;
    println!("tags: len={}", tags.len());

    store.with_map("metadata", |map| {
        map.insert("version", "0.1.0")?.insert("stable", false)?;
        Ok(())
    })?;

    let version = store.get("metadata")?.get_key("version")?.get_value()?;
    println!("metadata.version = {version:?}");

    // --- Nested construction via MapHandle closures ---
    // with_array and with_map on MapHandle let you build nested structures
    // without manually managing sub-handles.
    store.with_map("server", |map| {
        map.insert("name", "prod-01")?;

        map.with_array("ports", |arr| {
            arr.push(80)?.push(443)?;
            Ok(())
        })?;

        map.with_map("tls", |tls| {
            tls.insert("enabled", true)?
                .insert("cert_path", "/etc/ssl/cert.pem")?;
            Ok(())
        })?;

        Ok(())
    })?;

    let tls_enabled = store
        .get("server")?
        .get_key("tls")?
        .get_key("enabled")?
        .get_value()?;
    println!("server.tls.enabled = {tls_enabled:?}");

    // --- push_map on ArrayHandle ---
    // Builds an array of maps — the most common pattern for tabular results.
    let mut arr = store.define_array("users")?;
    {
        let mut user = arr.push_map()?;
        user.insert("name", "Alice")?.insert("role", "admin")?;
    }
    {
        let mut user = arr.push_map()?;
        user.insert("name", "Bob")?.insert("role", "viewer")?;
    }

    let bob_role = store
        .get("users")?
        .get_index(1)?
        .get_key("role")?
        .get_value()?;
    println!("users[1].role = {bob_role:?}");
    assert_eq!(bob_role, &Value::Text("viewer".into()));

    println!("store_collections: ok");
    Ok(())
}

impl<'a, T> ArrayHandle<'a, Vec<T>>

pub fn push_array(&mut self) -> Result<ArrayHandle<'_, T>, StoreError>

Appends a new inner Vec<T> and returns a child array handle for populating it.

impl<'a, T> ArrayHandle<'a, HashMap<String, T>>

pub fn push_map(&mut self) -> Result<MapHandle<'_, T>, StoreError>

Appends a new inner HashMap<String, T> and returns a child map handle for populating it.

impl<'a, T> ArrayHandle<'a, T>

pub fn new<N: Into<String>>(name: N, data: &'a mut Vec<T>) -> Self

Wraps a mutable vector in a handle under the given display name. Intended for extension code; user code obtains handles through Pipeline::array or Store::define_array.

pub fn push<V: Into<T>>(&mut self, value: V) -> Result<&mut Self, StoreError>

Appends a value to the array. Returns a mutable reference for chaining.

Examples found in repository

examples/extend_store_nested.rs (line 35)

fn write_grouped_data(store: &mut Store) -> Result<(), Box<dyn std::error::Error>> {
    store.with_map("by_severity", |groups| {
        groups.with_array("high", |arr| {
            arr.push("INC-001")?.push("INC-004")?;
            Ok(())
        })?;

        groups.with_array("medium", |arr| {
            arr.push("INC-002")?;
            Ok(())
        })?;

        groups.with_array("low", |arr| {
            arr.push("INC-003")?.push("INC-005")?.push("INC-006")?;
            Ok(())
        })?;

        Ok(())
    })?;

    Ok(())
}

examples/extend_store_collections.rs (line 15)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut store = Store::new();

    // --- Chainable push ---
    // push() accepts V: Into<StoreEntry>. The blanket impl
    // `From<T: Into<Value>> for StoreEntry` means primitives work directly.
    let mut arr = store.define_array("numbers")?.push(1)?.push(2)?.push(3)?;

    let items = store.get("numbers")?.as_array()?;
    println!("numbers: len={}", items.len());
    assert_eq!(items.len(), 3);

    // --- Chainable insert ---
    // insert() accepts K: Into<String> and V: Into<StoreEntry>.
    let mut map = store.define_map("config")?;
    map.insert("host", "localhost")?
        .insert("port", 8080i64)?
        .insert("debug", true)?;

    let host = store.get("config")?.get_key("host")?.get_value()?;
    println!("config.host = {host:?}");

    // --- Closure API on Store ---
    // with_array scopes the handle so you don't need an explicit drop.
    store.with_array("tags", |arr| {
        arr.push("rust")?.push("pipeline")?.push("store")?;
        Ok(())
    })?;

    let tags = store.get("tags")?.as_array()?;
    println!("tags: len={}", tags.len());

    store.with_map("metadata", |map| {
        map.insert("version", "0.1.0")?.insert("stable", false)?;
        Ok(())
    })?;

    let version = store.get("metadata")?.get_key("version")?.get_value()?;
    println!("metadata.version = {version:?}");

    // --- Nested construction via MapHandle closures ---
    // with_array and with_map on MapHandle let you build nested structures
    // without manually managing sub-handles.
    store.with_map("server", |map| {
        map.insert("name", "prod-01")?;

        map.with_array("ports", |arr| {
            arr.push(80)?.push(443)?;
            Ok(())
        })?;

        map.with_map("tls", |tls| {
            tls.insert("enabled", true)?
                .insert("cert_path", "/etc/ssl/cert.pem")?;
            Ok(())
        })?;

        Ok(())
    })?;

    let tls_enabled = store
        .get("server")?
        .get_key("tls")?
        .get_key("enabled")?
        .get_value()?;
    println!("server.tls.enabled = {tls_enabled:?}");

    // --- push_map on ArrayHandle ---
    // Builds an array of maps — the most common pattern for tabular results.
    let mut arr = store.define_array("users")?;
    {
        let mut user = arr.push_map()?;
        user.insert("name", "Alice")?.insert("role", "admin")?;
    }
    {
        let mut user = arr.push_map()?;
        user.insert("name", "Bob")?.insert("role", "viewer")?;
    }

    let bob_role = store
        .get("users")?
        .get_index(1)?
        .get_key("role")?
        .get_value()?;
    println!("users[1].role = {bob_role:?}");
    assert_eq!(bob_role, &Value::Text("viewer".into()));

    println!("store_collections: ok");
    Ok(())
}

examples/iter_array.rs (line 17)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // ── Basic iter_array: iterate over items and capture each ──
    println!("=== Basic iter_array ===");
    {
        let mut pipe = Pipeline::default();
        pipe.array("numbers")?.push(10)?.push(20)?.push(30)?;

        pipe.iter_array(
            "process",
            IterSource::array("numbers"),
            |_index, item, body| {
                body.step::<SetVar>(
                    "capture",
                    params!(
                        "name" => "doubled",
                        "value" => Param::reference(item),
                    ),
                )?;
                Ok(())
            },
        )?;

        pipe.hook(Logger::new().writer(std::io::stdout()));
        pipe.hook(Timeout::new(Duration::from_secs(5)));

        let complete = pipe.compile()?.run().wait()?;
        complete.debug();
    }

    // ── iter_array with parent variable references ──
    println!("\n=== iter_array referencing parent vars ===");
    {
        let mut pipe = Pipeline::default();
        pipe.var("prefix", "item_")?;
        pipe.array("items")?
            .push("alpha")?
            .push("beta")?
            .push("gamma")?;

        pipe.iter_array("loop", IterSource::array("items"), |_index, item, body| {
            body.step::<SetVar>(
                "combine",
                params!(
                    "name" => "label",
                    "value" => Param::template(vec![
                        Param::reference("prefix"),
                        Param::reference(item),
                    ]),
                ),
            )?;
            Ok(())
        })?;

        let complete = pipe.compile()?.run().wait()?;
        complete.debug();
    }

    // ── iter_array with multiple body steps ──
    println!("\n=== iter_array with multi-step body ===");
    {
        let mut pipe = Pipeline::default();
        pipe.var("suffix", "_processed")?;
        pipe.array("words")?.push("hello")?.push("world")?;

        pipe.iter_array("each", IterSource::array("words"), |_index, item, body| {
            body.step::<SetVar>(
                "tag",
                params!(
                    "name" => "tagged",
                    "value" => Param::template(vec![
                        Param::reference(item),
                        Param::reference("suffix"),
                    ]),
                ),
            )?;
            body.step::<SetVar>(
                "wrap",
                params!(
                    "name" => "wrapped",
                    "value" => Param::template(vec![
                        Param::literal("["),
                        Param::reference("tagged"),
                        Param::literal("]"),
                    ]),
                ),
            )?;
            Ok(())
        })?;

        let complete = pipe.compile()?.run().wait()?;
        complete.debug();
    }

    // ── Error: iter_array with unresolved source ──
    println!("\n=== iter_array unresolved source ===");
    {
        let mut pipe = Pipeline::default();

        pipe.iter_array("loop", IterSource::array("nonexistent"), |_, _, _| Ok(()))?;

        match pipe.compile() {
            Err(e) => println!("  Caught: {}", e),
            Ok(_) => println!("  ERROR: should have failed!"),
        }
    }

    // ── Error: iter_array body references unknown variable ──
    println!("\n=== iter_array unresolved body reference ===");
    {
        let mut pipe = Pipeline::default();
        pipe.array("items")?.push(1)?;

        pipe.iter_array("loop", IterSource::array("items"), |_, _, body| {
            body.step::<SetVar>(
                "bad",
                params!(
                    "name" => "out",
                    "value" => Param::reference("ghost"),
                ),
            )?;
            Ok(())
        })?;

        match pipe.compile() {
            Err(e) => println!("  Caught: {}", e),
            Ok(_) => println!("  ERROR: should have failed!"),
        }
    }

    Ok(())
}

examples/pipeline.rs (line 21)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // ── Happy path: derived outputs resolved at compile time ──
    println!("=== Valid pipeline ===");
    {
        let mut pipe = Pipeline::default();

        pipe.var("number", 1)?;
        pipe.var("string", "example")?;
        pipe.array("array")?.push(1)?.push(2)?.push(3)?;

        // Derived output name from a variable ref: "string" → "example"
        pipe.step::<SetVar>(
            "set_var_1",
            params!(
                "name" => Param::reference("string"),
                "value" => Param::reference("number"),
            ),
        )?;

        // Derived output name from a literal: "greeting"
        pipe.step::<SetVar>(
            "set_var_2",
            params!(
                "name" => "greeting",
                "value" => "hello world",
            ),
        )?;

        // Chained: use derived output "greeting" from set_var_2 as a variable reference
        pipe.step::<SetVar>(
            "set_var_3",
            params!(
                "name" => "farewell",
                "value" => Param::reference("greeting"),
            ),
        )?;

        // Template referencing a derived output
        pipe.step::<SetVar>(
            "set_var_4",
            params!(
                "name" => "message",
                "value" => Param::template(vec![
                    Param::reference("greeting"),
                    Param::literal(" and goodbye"),
                ]),
            ),
        )?;

        pipe.returns(
            "summary",
            params!(
                "original_number" => Param::reference("number"),
                "original_string" => Param::reference("string"),
                "the_array" => Param::reference("array"),
                "greeting" => Param::reference("greeting"),
                "derived_from_ref" => Param::reference("example"),
                "farewell" => Param::reference("farewell"),
                "message" => Param::reference("message"),
            ),
        )?;

        // Logger: writes each event to stdout
        pipe.hook(Logger::new().writer(std::io::stdout()));

        // Profiler: tracks per-step wall-clock duration
        let profiler = Profiler::new().writer(std::io::stdout());
        let timings = profiler.timings();
        pipe.hook(profiler);

        // EventLog: collects structured event records for post-mortem inspection
        let event_log = EventLog::new();
        let log = event_log.log();
        pipe.hook(event_log);

        // Timeout: aborts if pipeline exceeds 5 seconds
        pipe.hook(Timeout::new(Duration::from_secs(5)));

        // StoreValidator: assert "greeting" exists as Text after set_var_2
        pipe.hook(
            StoreValidator::new()
                .after_step("set_var_2")
                .expect("greeting", Type::Text),
        );

        let complete = pipe.compile()?.run().wait()?;
        complete.debug();

        // Read profiler timings programmatically
        println!("\nProgrammatic timings:");
        for (name, dur) in timings.lock().unwrap().iter() {
            println!("  {} took {:.3}ms", name, dur.as_secs_f64() * 1000.0);
        }

        // Read event log
        println!("\nEvent log ({} events):", log.lock().unwrap().len());
        for record in log.lock().unwrap().iter() {
            println!(
                "  [{:.3}ms] {:?}{}",
                record.elapsed.as_secs_f64() * 1000.0,
                record.kind,
                record
                    .step_name
                    .as_deref()
                    .map(|s| format!(" ({})", s))
                    .unwrap_or_default(),
            );
        }
    }

    // ── Error: step references a variable that doesn't exist ──
    println!("\n=== Unresolved step reference ===");
    {
        let mut pipe = Pipeline::default();
        pipe.var("number", 1)?;

        pipe.step::<SetVar>(
            "bad_step",
            params!(
                "name" => "output",
                "value" => Param::reference("does_not_exist"),
            ),
        )?;

        match pipe.compile() {
            Err(e) => println!("  Caught: {}", e),
            Ok(_) => println!("  ERROR: should have failed!"),
        }
    }

    // ── Error: return references something that was never produced ──
    println!("\n=== Unresolved return reference ===");
    {
        let mut pipe = Pipeline::default();
        pipe.var("number", 1)?;

        pipe.step::<SetVar>(
            "ok_step",
            params!(
                "name" => "output",
                "value" => Param::reference("number"),
            ),
        )?;

        pipe.returns(
            "result",
            params!(
                "good" => Param::reference("output"),
                "bad" => Param::reference("never_created"),
            ),
        )?;

        match pipe.compile() {
            Err(e) => println!("  Caught: {}", e),
            Ok(_) => println!("  ERROR: should have failed!"),
        }
    }

    // ── Error: step references a later step's output (ordering) ──
    println!("\n=== Forward reference (wrong order) ===");
    {
        let mut pipe = Pipeline::default();
        pipe.var("base", "start")?;

        // This step tries to reference "later_output" which doesn't exist yet
        pipe.step::<SetVar>(
            "step_1",
            params!(
                "name" => "early",
                "value" => Param::reference("later_output"),
            ),
        )?;

        pipe.step::<SetVar>(
            "step_2",
            params!(
                "name" => "later_output",
                "value" => Param::reference("base"),
            ),
        )?;

        match pipe.compile() {
            Err(e) => println!("  Caught: {}", e),
            Ok(_) => println!("  ERROR: should have failed!"),
        }
    }

    // ── Error: duplicate variable name ──
    println!("\n=== Duplicate variable ===");
    {
        let mut pipe = Pipeline::default();
        pipe.var("x", 1)?;
        match pipe.var("x", 2) {
            Err(e) => println!("  Caught: {}", e),
            Ok(_) => println!("  ERROR: should have failed!"),
        }
    }

    // ── Error: duplicate step name ──
    println!("\n=== Duplicate step ===");
    {
        let mut pipe = Pipeline::default();
        pipe.var("v", "val")?;

        pipe.step::<SetVar>("same_name", params!("name" => "a", "value" => "b"))?;
        match pipe.step::<SetVar>("same_name", params!("name" => "c", "value" => "d")) {
            Err(e) => println!("  Caught: {}", e),
            Ok(_) => println!("  ERROR: should have failed!"),
        }
    }

    // ── StepFilter: deny a step from executing ──
    println!("\n=== StepFilter deny ===");
    {
        let mut pipe = Pipeline::default();
        pipe.var("x", 1)?;

        pipe.step::<SetVar>("allowed_step", params!("name" => "a", "value" => "ok"))?;
        pipe.step::<SetVar>(
            "blocked_step",
            params!("name" => "b", "value" => Param::reference("a")),
        )?;

        pipe.returns("result", params!("a" => Param::reference("a")))?;

        pipe.hook(StepFilter::deny(["blocked_step"]));

        match pipe.compile()?.run().wait() {
            Err(e) => println!("  Caught: {}", e),
            Ok(_) => println!("  ERROR: should have been denied!"),
        }
    }

    Ok(())
}