Function call 

pub fn call<F: Function>(f: F, input: Value) -> Call<F>

One call of f, with its config compiled from input as a workflow would compile it; .await runs it.

The builder’s setters are optional: an empty message, no layers, ExprEngine::default, the name "function", changes captured, and no channel, deadline or cancellation token.

Panics

When awaited, wherever Registry::register panics: on a name it reserves, or on a spec the function’s config cannot honour.

use dataflow_rs::datavalue::OwnedDataValue;
use dataflow_rs::expr::Output;
use dataflow_rs::function::testing::{self, Write};
use dataflow_rs::function::{BoxFuture, CallLayer, FunctionConfig, Next};
use dataflow_rs::program::IssueCode;
use dataflow_rs::{Ctx, Expr, Function, Message, Outcome, Result};
use serde_json::json;
use std::collections::BTreeMap;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;

#[derive(FunctionConfig)]
struct DbRead {
    query: Expr<String>,
    #[output(default = "data.rows")]
    output: Output,
}

/// Answers each query from the rows it was built with, as the host's
/// database would.
struct MyFn {
    rows: BTreeMap<String, serde_json::Value>,
}

impl MyFn {
    fn new(rows: BTreeMap<String, serde_json::Value>) -> Self {
        Self { rows }
    }
}

impl Function for MyFn {
    type Config = DbRead;

    async fn call(&self, ctx: &mut Ctx<'_>, cfg: &DbRead) -> Result<Outcome> {
        let query = ctx.eval(&cfg.query)?;
        let Some(rows) = self.rows.get(&*query) else {
            return Ok(Outcome::Status(404));
        };
        ctx.emit(&cfg.output, rows.into())?;
        Ok(Outcome::Ok)
    }
}

/// Counts the calls it wraps.
struct Counted(Arc<AtomicUsize>);

impl CallLayer for Counted {
    fn call<'a>(&'a self, ctx: &'a mut Ctx<'_>, next: Next<'a>) -> BoxFuture<'a, Result<Outcome>> {
        self.0.fetch_add(1, Ordering::Relaxed);
        next.run(ctx)
    }
}

let deps = BTreeMap::from([(
    "select name from customers where id = 7".to_owned(),
    json!([{"name": "Ada"}]),
)]);
let calls = Arc::new(AtomicUsize::new(0));
let input = json!({
    "query": {"cat": ["select name from customers where id = ", {"var": "data.id"}]},
    "output": "data.rows",
});
let data = OwnedDataValue::from(&json!({"id": 7}));
let out = testing::call(MyFn::new(deps), input)
    .message(Message::from_parts(OwnedDataValue::Null, data, OwnedDataValue::Null))
    .layer(Counted(Arc::clone(&calls)))
    .await;

let rows = OwnedDataValue::from(&json!([{"name": "Ada"}]));
assert!(out.issues.is_empty());
assert_eq!(out.outcome.unwrap(), Outcome::Ok);
assert_eq!(out.writes, [("data.rows".parse().unwrap(), Write::Set(rows.clone()))]);
assert_eq!(out.changes.len(), 1);
assert_eq!(out.message.context().get(&"data.rows".parse().unwrap()), Some(&rows));
assert_eq!(calls.load(Ordering::Relaxed), 1);

// A config alone; `Err` carries every issue.
let cfg: DbRead = testing::config(json!({"query": "select 1"})).ok().unwrap();
assert_eq!(cfg.output.constant().unwrap().to_string(), "data.rows");
let issues = testing::config::<DbRead>(json!({"qeury": "select 1"})).err().unwrap();
let codes: Vec<IssueCode> = issues.iter().map(|i| i.code).collect();
assert_eq!(codes, [IssueCode::LikelyTypo, IssueCode::InputParse]);