dataflow_rs/function/
mod.rs

1//! Functions: the typed units of work a task runs.
2
3pub(crate) mod builtins;
4mod call;
5mod config;
6mod ctx;
7mod erased;
8mod layer;
9pub mod layers;
10mod outcome;
11mod overlay;
12mod registry;
13mod spec;
14mod sync;
15mod templated;
16pub mod testing;
17
18pub use crate::expr::{ExprVisitor, FieldKind, FieldSpec, Rendered, ValueKind};
19pub use call::{Function, Prepare};
20pub use config::FunctionConfig;
21pub(crate) use ctx::{Binding, CallStart, Primary};
22pub use ctx::{CallInfo, Ctx};
23/// Derives [`FunctionConfig`](trait@FunctionConfig) (and `ConfigField`, so a
24/// derived config nests).
25pub use dataflow_rs_derive::FunctionConfig;
26pub(crate) use erased::BoundCall;
27pub use layer::{BoxFuture, CallLayer, Next};
28pub use outcome::Outcome;
29pub(crate) use overlay::Overlay;
30pub(crate) use registry::Entry;
31pub use registry::{FunctionInfo, Registry};
32pub use spec::{FunctionSpec, Reference, RetrySafety};
33pub(crate) use sync::{SyncCtx, eval, task_record, test};
34pub use templated::Templated;
35
36/// What `#[derive(FunctionConfig)]` expands to names, and nothing else. Not an
37/// API: the derive emits `::dataflow_rs::function::__private::…` paths, which
38/// resolve here.
39#[doc(hidden)]
40pub mod __private {
41    pub use super::config::FunctionConfig;
42    pub use crate::base::{AuthoredPath, Result};
43    pub use crate::expr::{
44        CompileCx, ConfigField, ExprVisitor, FieldKind, FieldSpec, Output, Resolver, ValueKind,
45        Visit,
46    };
47    pub use datavalue::OwnedDataValue;
48    pub use serde_json::Value;
49    pub use std::sync::OnceLock;
50
51    use crate::base::{Error, ErrorKind, IssueCode};
52    use serde::Serialize;
53    use serde::de::DeserializeOwned;
54    use serde_json::Map;
55    use std::borrow::Cow;
56
57    /// One field's key and its declared default, for the compile helpers.
58    #[derive(Clone, Copy, Debug)]
59    pub struct Key {
60        /// The authored key.
61        key: &'static str,
62        /// The default, as JSON text.
63        default: Option<&'static str>,
64        /// An output: an explicit `null` is absent.
65        output: bool,
66    }
67
68    impl Key {
69        /// A field's key: its authored name, its default as JSON text, and
70        /// whether it is an output.
71        pub const fn new(key: &'static str, default: Option<&'static str>, output: bool) -> Self {
72            Self {
73                key,
74                default,
75                output,
76            }
77        }
78    }
79
80    /// The input as an object, or `INPUT_PARSE` at the current path.
81    pub fn object<'v>(input: &'v Value, cx: &mut CompileCx<'_>) -> Option<&'v Map<String, Value>> {
82        let object = input.as_object();
83        if object.is_none() {
84            cx.issue(IssueCode::InputParse, "expected an object");
85        }
86        object
87    }
88
89    /// Reports each key of `object` that is not in `known`, in the input's
90    /// key order.
91    pub fn unknown_keys(object: &Map<String, Value>, known: &[&str], cx: &mut CompileCx<'_>) {
92        for key in object.keys().filter(|key| !known.contains(&key.as_str())) {
93            cx.at(key.as_str(), |cx| cx.unknown_key(key, known));
94        }
95    }
96
97    /// The field's raw value: present (and, for an output, not `null`), else
98    /// its parsed default.
99    fn raw(object: &Map<String, Value>, key: Key) -> Option<Cow<'_, Value>> {
100        match object.get(key.key) {
101            Some(raw) if !(key.output && raw.is_null()) => Some(Cow::Borrowed(raw)),
102            _ => key.default.map(|text| Cow::Owned(default(text, key.key))),
103        }
104    }
105
106    /// A declared default, parsed. The derive writes every default as JSON
107    /// and refuses a `default_json` that is not, so this cannot fail.
108    fn default(text: &str, key: &str) -> Value {
109        serde_json::from_str(text).unwrap_or_else(|e| {
110            unreachable!("the default of `{key}` was validated by the derive ({e}): {text}")
111        })
112    }
113
114    /// `INPUT_PARSE` for a required field the input leaves out.
115    pub(crate) fn missing(key: &str, cx: &mut CompileCx<'_>) {
116        cx.issue(IssueCode::InputParse, format!("missing field `{key}`"));
117    }
118
119    /// Compiles one field through its [`ConfigField`] impl, at its key.
120    pub fn field<T: ConfigField>(
121        object: &Map<String, Value>,
122        key: Key,
123        cx: &mut CompileCx<'_>,
124    ) -> Option<T> {
125        cx.at(key.key, |cx| match raw(object, key) {
126            Some(raw) => T::compile(&raw, cx),
127            None => T::absent().or_else(|| {
128                missing(key.key, cx);
129                None
130            }),
131        })
132    }
133
134    /// Compiles one `#[config(serde)]` field with `serde_json`, at its key.
135    pub fn serde_field<T: DeserializeOwned>(
136        object: &Map<String, Value>,
137        key: Key,
138        cx: &mut CompileCx<'_>,
139    ) -> Option<T> {
140        cx.at(key.key, |cx| {
141            let Some(raw) = raw(object, key) else {
142                return T::deserialize(&Value::Null).map_or_else(
143                    |_| {
144                        missing(key.key, cx);
145                        None
146                    },
147                    Some,
148                );
149            };
150            T::deserialize(&*raw)
151                .map_err(|e| cx.issue(IssueCode::InputParse, e.to_string()))
152                .ok()
153        })
154    }
155
156    /// Whether a `#[config(serde)]` field is required: `T` cannot be read
157    /// from `null`.
158    pub fn serde_required<T: DeserializeOwned>() -> bool {
159        T::deserialize(&Value::Null).is_err()
160    }
161
162    /// A resolved config's entries, in declaration order.
163    pub type Entries = Vec<(String, OwnedDataValue)>;
164
165    /// Resolves one field through its [`ConfigField`] impl, and keeps it at
166    /// `key` unless it is unset.
167    ///
168    /// # Errors
169    ///
170    /// The field's.
171    pub fn resolve_field<T: ConfigField>(
172        entries: &mut Entries,
173        key: &str,
174        field: &T,
175        r: &mut dyn Resolver,
176    ) -> Result<()> {
177        if let Some(value) = field.resolve(r)? {
178            entries.push((key.to_owned(), value));
179        }
180        Ok(())
181    }
182
183    /// Resolves one `#[config(serde)]` field as it serializes, and keeps it
184    /// at `key` unless that is `null`. The JSON text is read back, so a
185    /// struct's or a map's keys keep the order its `Serialize` writes them
186    /// in.
187    ///
188    /// # Errors
189    ///
190    /// [`ErrorKind::Config`] when the value does not serialize.
191    pub fn resolve_serde<T: Serialize>(entries: &mut Entries, key: &str, field: &T) -> Result<()> {
192        let refused = |e: serde_json::Error| {
193            Error::new(
194                ErrorKind::Config,
195                format!("the field `{key}` does not serialize: {e}"),
196            )
197        };
198        let text = serde_json::to_string(field).map_err(refused)?;
199        // Read as a message's values are, so a whole-number float stays a
200        // float, as the JSON bridge keeps one.
201        let value =
202            crate::base::exact(&mut serde_json::Deserializer::from_str(&text)).map_err(refused)?;
203        if !value.is_null() {
204            entries.push((key.to_owned(), value));
205        }
206        Ok(())
207    }
208
209    /// The assembled config, unless this compile recorded a Rejected issue.
210    pub fn finish<T>(out: T, rejected_before: usize, cx: &CompileCx<'_>) -> Option<T> {
211        (cx.rejected() == rejected_before).then_some(out)
212    }
213
214    /// One [`FieldSpec`], as the derive declares it. A field holding an
215    /// output records whether or not it says so.
216    pub fn spec(
217        name: &'static str,
218        kind: FieldKind,
219        required: bool,
220        records: bool,
221        default: Option<&'static str>,
222        primary: bool,
223        description: Option<&'static str>,
224    ) -> FieldSpec {
225        FieldSpec {
226            name: Cow::Borrowed(name),
227            kind,
228            required,
229            records: records || kind.contains_output(),
230            default: default.map(Cow::Borrowed),
231            primary,
232            description: description.map(Cow::Borrowed),
233        }
234    }
235}
dataflow_rs/function/mod.rs

dataflow_rs/function/
mod.rs
