dataflow_rs/function/
call.rs

1//! `Function`: the trait a host function implements, and what `prepare`
2//! is told.
3
4use super::{Ctx, FunctionConfig, FunctionSpec, Outcome};
5use crate::base::{Issue, Result};
6
7/// A host function: a typed config compiled once at build, and a call per
8/// task run.
9///
10/// `call` returns a `Send` future, so an implementation may be written with
11/// `async fn`. [`testing::call`](super::testing::call) runs one call of it
12/// outside any workflow:
13///
14/// ```
15/// # tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
16/// use dataflow_rs::datavalue::OwnedDataValue;
17/// use dataflow_rs::expr::Output;
18/// use dataflow_rs::function::{FunctionConfig, FunctionSpec, testing};
19/// use dataflow_rs::{Ctx, Expr, Function, Outcome, Registry, Result};
20/// use serde_json::json;
21///
22/// #[derive(FunctionConfig)]
23/// struct FetchConfig {
24///     url: Expr<String>,
25///     #[output(default = "data.response")]
26///     output: Output,
27/// }
28///
29/// struct Fetch;
30///
31/// impl Function for Fetch {
32///     type Config = FetchConfig;
33///
34///     fn spec(&self) -> FunctionSpec {
35///         FunctionSpec::default().with_category("io")
36///     }
37///
38///     async fn call(&self, ctx: &mut Ctx<'_>, cfg: &FetchConfig) -> Result<Outcome> {
39///         let url = ctx.eval(&cfg.url)?;
40///         // A real fetch awaits its response here.
41///         let response = OwnedDataValue::from(&json!({"url": &*url, "status": 200}));
42///         ctx.emit(&cfg.output, response)?;
43///         Ok(Outcome::Ok)
44///     }
45/// }
46///
47/// let mut registry = Registry::default();
48/// registry.register("fetch", Fetch);
49///
50/// let called = testing::call(Fetch, json!({"url": "https://example.com"})).await;
51/// assert_eq!(called.outcome.unwrap(), Outcome::Ok);
52/// let status = called.message.context().get(&"data.response.status".parse().unwrap());
53/// assert_eq!(status, Some(&OwnedDataValue::from(&json!(200))));
54/// # });
55/// ```
56pub trait Function: Send + Sync + 'static {
57    /// The function's `input`, compiled at build.
58    type Config: FunctionConfig;
59
60    /// What the function says about itself, for tooling. Called once, when
61    /// it is registered.
62    fn spec(&self) -> FunctionSpec {
63        FunctionSpec::default()
64    }
65
66    /// Checks or completes a compiled config at build, with what the config
67    /// alone cannot know. It runs only when the config compiled.
68    ///
69    /// Issue paths are relative to the config. The compiler prefixes the
70    /// task's `function.input` path and stamps the step, and a Rejected
71    /// issue fails the task's compile.
72    fn prepare(
73        &self,
74        _cfg: &mut Self::Config,
75        _p: &Prepare<'_>,
76    ) -> std::result::Result<(), Vec<Issue>> {
77        Ok(())
78    }
79
80    /// Runs one call.
81    fn call(
82        &self,
83        ctx: &mut Ctx<'_>,
84        cfg: &Self::Config,
85    ) -> impl Future<Output = Result<Outcome>> + Send;
86}
87
88/// What [`Function::prepare`] is told about the call site it prepares.
89///
90/// Its fields are private, behind getters, so it can grow.
91#[derive(Debug)]
92pub struct Prepare<'a> {
93    name: &'a str,
94    workflow_id: &'a str,
95    task_id: &'a str,
96    spec: &'a FunctionSpec,
97}
98
99impl<'a> Prepare<'a> {
100    /// What `prepare` is told about the task `task_id` of the workflow
101    /// `workflow_id`, which calls the function registered as `name` with
102    /// `spec`.
103    pub(crate) fn new(
104        name: &'a str,
105        workflow_id: &'a str,
106        task_id: &'a str,
107        spec: &'a FunctionSpec,
108    ) -> Self {
109        Self {
110            name,
111            workflow_id,
112            task_id,
113            spec,
114        }
115    }
116
117    /// The name the function is registered under.
118    pub fn name(&self) -> &'a str {
119        self.name
120    }
121
122    /// The id of the workflow the call site is in.
123    pub fn workflow_id(&self) -> &'a str {
124        self.workflow_id
125    }
126
127    /// The id of the task that calls the function.
128    pub fn task_id(&self) -> &'a str {
129        self.task_id
130    }
131
132    /// The function's spec, as it was when registered.
133    pub fn spec(&self) -> &'a FunctionSpec {
134        self.spec
135    }
136}
dataflow_rs/function/call.rs

dataflow_rs/function/
call.rs
