dataflow_rs/function/
layers.rs

1//! Ready-made [`CallLayer`]s.
2
3use super::{BoxFuture, CallLayer, Ctx, Next, Outcome};
4use crate::base::Result;
5use datavalue::OwnedDataValue;
6use std::fmt;
7
8/// What an [`Answer`] asks of each call.
9type Answering = dyn Fn(&Ctx<'_>) -> Option<OwnedDataValue> + Send + Sync;
10
11/// Answers calls from a closure, for a dry run.
12///
13/// The closure is asked about each call it wraps, and sees the call's
14/// [`Ctx`]: its [`info`](Ctx::info), its [`input`](Ctx::input), resolved with
15/// every secret-reading field `null`, and the
16/// [`primary_output`](Ctx::primary_output) an answer is written through.
17/// `Some(value)` is written with [`Ctx::emit_default`], through the callee's
18/// primary output, and the call returns [`Outcome::Ok`] without reaching the
19/// layers inside this one or the function. `None` passes the call on. A
20/// callee with no primary output fails the call with a `Config` error.
21///
22/// ```
23/// use dataflow_rs::datavalue::OwnedDataValue;
24/// use dataflow_rs::function::{CallLayer, layers::Answer};
25/// use dataflow_rs::RunOptions;
26/// use std::sync::Arc;
27///
28/// // Every `http_call` that has somewhere to write answers with the input
29/// // it was given; the rest run.
30/// let dry_run: Vec<Arc<dyn CallLayer>> = vec![Arc::new(Answer::new(|ctx| {
31///     if ctx.info().function != "http_call" || ctx.primary_output().is_none() {
32///         return None;
33///     }
34///     let input = ctx.input().ok()?;
35///     Some(OwnedDataValue::Object(vec![
36///         ("status".into(), "dry run".into()),
37///         ("request".into(), input),
38///     ]))
39/// }))];
40/// let opts = RunOptions::default().layers(&dry_run);
41/// ```
42pub struct Answer {
43    answer: Box<Answering>,
44}
45
46impl Answer {
47    /// A layer answering each call `f` gives a value for.
48    pub fn new(f: impl Fn(&Ctx<'_>) -> Option<OwnedDataValue> + Send + Sync + 'static) -> Self {
49        Self {
50            answer: Box::new(f),
51        }
52    }
53}
54
55impl CallLayer for Answer {
56    fn call<'a>(&'a self, ctx: &'a mut Ctx<'_>, next: Next<'a>) -> BoxFuture<'a, Result<Outcome>> {
57        match (self.answer)(ctx) {
58            Some(value) => Box::pin(std::future::ready(
59                ctx.emit_default(value).map(|()| Outcome::Ok),
60            )),
61            None => next.run(ctx),
62        }
63    }
64}
65
66impl fmt::Debug for Answer {
67    /// Nothing of the closure.
68    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
69        f.debug_struct("Answer").finish_non_exhaustive()
70    }
71}
dataflow_rs/function/layers.rs

dataflow_rs/function/
layers.rs
