use crate::engine::error::Result;
use crate::engine::executor::ArenaContext;
use crate::engine::functions::filter::FilterConfig;
use crate::engine::functions::integration::{EnrichConfig, HttpCallConfig, PublishKafkaConfig};
use crate::engine::functions::log::LogConfig;
use crate::engine::functions::map::MapConfig;
use crate::engine::functions::parse::{ParseConfig, execute_parse_json_in_arena, parse_xml_in};
use crate::engine::functions::path_template::ParamCtx;
use crate::engine::functions::publish::{PublishConfig, publish_json_in, publish_xml_in};
use crate::engine::functions::template::Template;
use crate::engine::functions::validation::ValidationConfig;
use crate::engine::message::{Change, Message};
use crate::engine::task_outcome::TaskOutcome;
use datalogic_rs::Engine;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Deserializer};
use serde_json::Value;
use std::any::Any;
use std::sync::Arc;
#[derive(Clone)]
pub struct CompiledCustomInput(pub Arc<dyn Any + Send + Sync>);
impl CompiledCustomInput {
#[inline]
pub fn as_any(&self) -> &(dyn Any + Send + Sync) {
&*self.0
}
}
impl std::fmt::Debug for CompiledCustomInput {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("CompiledCustomInput(<opaque>)")
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConnectorName<'a> {
Static(&'a str),
Computed(&'a Value),
}
impl<'a> ConnectorName<'a> {
fn of(template: &'a Template) -> Self {
match template.as_json() {
Value::String(s) => Self::Static(s),
other => Self::Computed(other),
}
}
pub fn as_static(&self) -> Option<&'a str> {
match self {
Self::Static(s) => Some(s),
Self::Computed(_) => None,
}
}
}
#[derive(Debug, Clone)]
pub enum FunctionConfig {
Map {
name: MapName,
input: MapConfig,
},
Validation {
name: ValidationName,
input: ValidationConfig,
},
ParseJson {
name: ParseJsonName,
input: ParseConfig,
},
ParseXml {
name: ParseXmlName,
input: ParseConfig,
},
PublishJson {
name: PublishJsonName,
input: PublishConfig,
},
PublishXml {
name: PublishXmlName,
input: PublishConfig,
},
Filter {
name: FilterName,
input: FilterConfig,
},
Log {
name: LogName,
input: LogConfig,
},
HttpCall {
name: HttpCallName,
input: HttpCallConfig,
},
Enrich {
name: EnrichName,
input: EnrichConfig,
},
PublishKafka {
name: PublishKafkaName,
input: PublishKafkaConfig,
},
Custom {
name: String,
input: Value,
compiled_input: Option<CompiledCustomInput>,
},
}
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum MapName {
Map,
}
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[serde(rename_all = "lowercase")]
pub enum ValidationName {
Validation,
Validate,
}
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum ParseJsonName {
ParseJson,
}
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum ParseXmlName {
ParseXml,
}
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum PublishJsonName {
PublishJson,
}
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum PublishXmlName {
PublishXml,
}
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[serde(rename_all = "lowercase")]
pub enum FilterName {
Filter,
}
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[serde(rename_all = "lowercase")]
pub enum LogName {
Log,
}
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum HttpCallName {
HttpCall,
}
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum EnrichName {
Enrich,
}
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum PublishKafkaName {
PublishKafka,
}
pub const BUILTIN_FUNCTION_NAMES: &[&str] = &[
"map",
"validation",
"validate",
"parse_json",
"parse_xml",
"publish_json",
"publish_xml",
"filter",
"log",
"http_call",
"enrich",
"publish_kafka",
];
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BuiltinKind {
SelfContained,
RequiresHandler,
}
pub fn builtin_function_kind(name: &str) -> Option<BuiltinKind> {
match name {
"map" | "validation" | "validate" | "parse_json" | "parse_xml" | "publish_json"
| "publish_xml" | "filter" | "log" => Some(BuiltinKind::SelfContained),
"http_call" | "enrich" | "publish_kafka" => Some(BuiltinKind::RequiresHandler),
_ => None,
}
}
#[inline]
pub fn is_builtin_function(name: &str) -> bool {
builtin_function_kind(name).is_some()
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DispatchableFunction<'a> {
pub name: &'a str,
pub kind: Option<BuiltinKind>,
pub aliases: &'static [&'static str],
}
const VALIDATE_ALIASES: &[&str] = &["validation"];
const NO_ALIASES: &[&str] = &[];
pub(crate) fn canonical_builtin_name(name: &str) -> &str {
match name {
"validation" => "validate",
other => other,
}
}
pub(crate) fn builtin_aliases(canonical: &str) -> &'static [&'static str] {
match canonical {
"validate" => VALIDATE_ALIASES,
_ => NO_ALIASES,
}
}
pub(crate) fn can_dispatch_in<V>(
registry: &std::collections::HashMap<String, V>,
name: &str,
) -> bool {
match builtin_function_kind(name) {
Some(BuiltinKind::SelfContained) => true,
_ => registry.contains_key(name),
}
}
pub(crate) fn dispatchable_functions_in<V>(
registry: &std::collections::HashMap<String, V>,
) -> impl Iterator<Item = DispatchableFunction<'_>> {
let builtins = BUILTIN_FUNCTION_NAMES
.iter()
.copied()
.filter(|name| canonical_builtin_name(name) == *name)
.filter_map(move |name| match builtin_function_kind(name) {
kind @ Some(BuiltinKind::SelfContained) => Some(DispatchableFunction {
name,
kind,
aliases: builtin_aliases(name),
}),
kind @ Some(BuiltinKind::RequiresHandler) if registry.contains_key(name) => {
Some(DispatchableFunction {
name,
kind,
aliases: builtin_aliases(name),
})
}
_ => None,
});
let customs = registry
.keys()
.map(String::as_str)
.filter(|name| builtin_function_kind(name).is_none())
.map(|name| DispatchableFunction {
name,
kind: None,
aliases: NO_ALIASES,
});
builtins.chain(customs)
}
fn parse_function_input<T, E>(func: &str, input: Value) -> std::result::Result<T, E>
where
T: DeserializeOwned,
E: serde::de::Error,
{
serde_json::from_value::<T>(input).map_err(|err| {
let raw = err.to_string();
let trimmed = raw
.rsplit_once(" at line ")
.map(|(head, _)| head)
.unwrap_or(&raw);
E::custom(format!("config for function '{func}': {trimmed}"))
})
}
impl<'de> Deserialize<'de> for FunctionConfig {
fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
where
D: Deserializer<'de>,
{
#[derive(Deserialize)]
struct Raw {
name: String,
input: Value,
}
let Raw { name, input } = Raw::deserialize(deserializer)?;
Ok(match name.as_str() {
"map" => Self::Map {
name: MapName::Map,
input: parse_function_input("map", input)?,
},
"validate" => Self::Validation {
name: ValidationName::Validate,
input: parse_function_input("validate", input)?,
},
"validation" => Self::Validation {
name: ValidationName::Validation,
input: parse_function_input("validation", input)?,
},
"parse_json" => Self::ParseJson {
name: ParseJsonName::ParseJson,
input: parse_function_input("parse_json", input)?,
},
"parse_xml" => Self::ParseXml {
name: ParseXmlName::ParseXml,
input: parse_function_input("parse_xml", input)?,
},
"publish_json" => Self::PublishJson {
name: PublishJsonName::PublishJson,
input: parse_function_input("publish_json", input)?,
},
"publish_xml" => Self::PublishXml {
name: PublishXmlName::PublishXml,
input: parse_function_input("publish_xml", input)?,
},
"filter" => Self::Filter {
name: FilterName::Filter,
input: parse_function_input("filter", input)?,
},
"log" => Self::Log {
name: LogName::Log,
input: parse_function_input("log", input)?,
},
"http_call" => Self::HttpCall {
name: HttpCallName::HttpCall,
input: parse_function_input("http_call", input)?,
},
"enrich" => Self::Enrich {
name: EnrichName::Enrich,
input: parse_function_input("enrich", input)?,
},
"publish_kafka" => Self::PublishKafka {
name: PublishKafkaName::PublishKafka,
input: parse_function_input("publish_kafka", input)?,
},
_ => Self::Custom {
name,
input,
compiled_input: None,
},
})
}
}
fn refresh_data_on_success(
message: &Message,
arena_ctx: &mut ArenaContext<'_>,
result: Result<(TaskOutcome, Vec<Change>)>,
) -> Result<(TaskOutcome, Vec<Change>)> {
if result.is_ok() {
arena_ctx.refresh_for_path(&message.context, "data");
}
result
}
impl FunctionConfig {
pub fn function_name(&self) -> &str {
match self {
Self::Map { .. } => "map",
Self::Validation { .. } => "validate",
Self::ParseJson { .. } => "parse_json",
Self::ParseXml { .. } => "parse_xml",
Self::PublishJson { .. } => "publish_json",
Self::PublishXml { .. } => "publish_xml",
Self::Filter { .. } => "filter",
Self::Log { .. } => "log",
Self::HttpCall { .. } => "http_call",
Self::Enrich { .. } => "enrich",
Self::PublishKafka { .. } => "publish_kafka",
Self::Custom { name, .. } => name,
}
}
pub fn connector(&self) -> Option<ConnectorName<'_>> {
match self {
Self::HttpCall { input, .. } => Some(ConnectorName::of(&input.connector)),
Self::Enrich { input, .. } => Some(ConnectorName::of(&input.connector)),
Self::PublishKafka { input, .. } => Some(ConnectorName::of(&input.connector)),
Self::Custom { input, .. } => input
.get("connector")
.and_then(Value::as_str)
.map(ConnectorName::Static),
Self::Map { .. }
| Self::Validation { .. }
| Self::ParseJson { .. }
| Self::ParseXml { .. }
| Self::PublishJson { .. }
| Self::PublishXml { .. }
| Self::Filter { .. }
| Self::Log { .. } => None,
}
}
pub fn is_sync_builtin(&self) -> bool {
!matches!(
self,
Self::HttpCall { .. }
| Self::Enrich { .. }
| Self::PublishKafka { .. }
| Self::Custom { .. }
)
}
pub(crate) fn try_execute_in_arena<'arena>(
&'arena self,
message: &mut Message,
arena_ctx: &mut ArenaContext<'arena>,
engine: &Arc<Engine>,
mapping_snapshots: Option<&mut Vec<Value>>,
) -> Option<Result<(TaskOutcome, Vec<Change>)>> {
match self {
Self::Map { input, .. } => {
Some(input.execute_in_arena(message, arena_ctx, engine, mapping_snapshots))
}
Self::Validation { input, .. } => {
Some(input.execute_in_arena(message, arena_ctx, engine))
}
Self::ParseJson { input, .. } => Some(execute_parse_json_in_arena(
message, input, engine, arena_ctx,
)),
Self::ParseXml { input, .. } => {
let p = ParamCtx::from_arena(engine, arena_ctx);
let result = parse_xml_in(message, input, p);
Some(refresh_data_on_success(message, arena_ctx, result))
}
Self::PublishJson { input, .. } => {
let p = ParamCtx::from_arena(engine, arena_ctx);
let result = publish_json_in(message, input, p);
Some(refresh_data_on_success(message, arena_ctx, result))
}
Self::PublishXml { input, .. } => {
let p = ParamCtx::from_arena(engine, arena_ctx);
let result = publish_xml_in(message, input, p);
Some(refresh_data_on_success(message, arena_ctx, result))
}
Self::Filter { input, .. } => Some(input.execute_in_arena(message, arena_ctx, engine)),
Self::Log { input, .. } => Some(input.execute_in_arena(message, arena_ctx, engine)),
Self::HttpCall { .. }
| Self::Enrich { .. }
| Self::PublishKafka { .. }
| Self::Custom { .. } => None,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn parse(value: serde_json::Value) -> std::result::Result<FunctionConfig, serde_json::Error> {
serde_json::from_value(value)
}
fn minimal_input(name: &str) -> serde_json::Value {
match name {
"map" => json!({ "mappings": [] }),
"validation" | "validate" => json!({ "rules": [] }),
"parse_json" | "parse_xml" | "publish_json" | "publish_xml" => {
json!({ "source": "data.in", "target": "out" })
}
"filter" => json!({ "condition": true }),
"log" => json!({ "message": "hi" }),
"http_call" => json!({ "connector": "c" }),
"enrich" => json!({ "connector": "c", "merge_path": "data.out" }),
"publish_kafka" => json!({ "connector": "c", "topic": "t" }),
_ => json!({}),
}
}
#[test]
fn map_with_valid_config_deserializes_to_map_variant() {
let cfg = parse(json!({
"name": "map",
"input": {
"mappings": [
{ "path": "data.x", "logic": { "var": "data.y" } }
]
}
}))
.expect("valid map config should deserialize");
assert!(matches!(cfg, FunctionConfig::Map { .. }));
}
#[test]
fn map_with_missing_mappings_gives_clear_error() {
let err = parse(json!({
"name": "map",
"input": {}
}))
.expect_err("map with empty input should fail");
let msg = err.to_string();
assert!(
msg.starts_with("config for function 'map':"),
"error should be prefixed with function envelope, got: {msg}"
);
assert!(
msg.contains("mappings"),
"error should mention the missing field, got: {msg}"
);
}
#[test]
fn map_with_wrong_input_shape_gives_clear_error() {
let err = parse(json!({
"name": "map",
"input": { "mappings": "not an array" }
}))
.expect_err("map with bad mappings type should fail");
let msg = err.to_string();
assert!(
msg.starts_with("config for function 'map':"),
"error should be prefixed with function envelope, got: {msg}"
);
}
#[test]
fn validation_accepts_both_spellings() {
for name in ["validate", "validation"] {
let cfg = parse(json!({
"name": name,
"input": { "rules": [] }
}))
.unwrap_or_else(|e| panic!("'{name}' should deserialize: {e}"));
assert!(matches!(cfg, FunctionConfig::Validation { .. }));
}
}
#[test]
fn unknown_name_falls_through_to_custom() {
let cfg = parse(json!({
"name": "my_custom_handler",
"input": { "anything": "goes" }
}))
.expect("unknown name should produce Custom");
match cfg {
FunctionConfig::Custom {
name,
compiled_input,
..
} => {
assert_eq!(name, "my_custom_handler");
assert!(compiled_input.is_none());
}
other => panic!("expected Custom, got {other:?}"),
}
}
#[test]
fn missing_name_field_errors() {
let err = parse(json!({ "input": {} })).expect_err("missing name should fail");
assert!(err.to_string().contains("name"));
}
#[test]
fn missing_input_field_errors() {
let err = parse(json!({ "name": "map" })).expect_err("missing input should fail");
assert!(err.to_string().contains("input"));
}
#[test]
fn http_call_with_missing_connector_gives_clear_error() {
let err = parse(json!({
"name": "http_call",
"input": { "method": "GET" }
}))
.expect_err("http_call needs connector");
let msg = err.to_string();
assert!(
msg.starts_with("config for function 'http_call':"),
"error should be prefixed with function envelope, got: {msg}"
);
assert!(msg.contains("connector"));
}
#[test]
fn builtin_names_never_fall_through_to_custom() {
for name in BUILTIN_FUNCTION_NAMES {
let cfg = parse(json!({
"name": name,
"input": {}
}));
match cfg {
Ok(c) => assert!(
!matches!(c, FunctionConfig::Custom { .. }),
"name '{name}' silently fell through to Custom"
),
Err(e) => assert!(
e.to_string()
.starts_with(&format!("config for function '{name}':")),
"name '{name}' failed without envelope: {e}"
),
}
assert!(
builtin_function_kind(name).is_some(),
"name '{name}' is in BUILTIN_FUNCTION_NAMES but classifies as None"
);
}
}
fn parse_http_call(
input: serde_json::Value,
) -> std::result::Result<HttpCallConfig, serde_json::Error> {
match parse(json!({ "name": "http_call", "input": input }))? {
FunctionConfig::HttpCall { input, .. } => Ok(input),
other => panic!("expected HttpCall, got {other:?}"),
}
}
#[test]
fn http_call_response_path_is_read_under_its_own_name() {
let cfg = parse_http_call(json!({ "connector": "c", "response_path": "data.x" }))
.expect("response_path should parse");
assert_eq!(
cfg.response_path.as_ref().map(Template::as_json),
Some(&json!("data.x"))
);
}
#[test]
fn http_call_response_path_accepts_the_output_alias() {
let cfg = parse_http_call(json!({ "connector": "c", "output": "data.x" }))
.expect("output should be accepted as an alias");
assert_eq!(
cfg.response_path.as_ref().map(Template::as_json),
Some(&json!("data.x"))
);
}
#[test]
fn http_call_response_path_is_optional() {
let cfg = parse_http_call(json!({ "connector": "c" })).expect("no destination is valid");
assert!(cfg.response_path.is_none());
}
#[test]
fn http_call_rejects_both_destination_keys_in_either_order() {
for input in [
json!({ "connector": "c", "response_path": "a", "output": "b" }),
json!({ "connector": "c", "output": "b", "response_path": "a" }),
] {
let err = parse_http_call(input.clone())
.expect_err("supplying both destination keys must fail");
let msg = err.to_string();
assert!(
msg.starts_with("config for function 'http_call':"),
"error should carry the function envelope, got: {msg}"
);
assert!(
msg.contains("duplicate field"),
"error should name the conflict, got: {msg}"
);
}
}
#[test]
fn http_call_rejects_a_misspelled_destination_field() {
for bad in ["outputs", "Output", "respose_path", "response-path"] {
let mut input = serde_json::Map::new();
input.insert("connector".to_string(), json!("c"));
input.insert(bad.to_string(), json!("data.x"));
let err = parse_http_call(serde_json::Value::Object(input))
.expect_err("a misspelled field must be rejected, not silently discarded");
let msg = err.to_string();
assert!(
msg.starts_with("config for function 'http_call':"),
"error should carry the function envelope, got: {msg}"
);
assert!(
msg.contains("unknown field"),
"error should say the field is unknown, got: {msg}"
);
assert!(
msg.contains(bad),
"error should name the offending field '{bad}', got: {msg}"
);
}
}
#[test]
fn enrich_does_not_accept_the_output_alias() {
let err = parse(json!({
"name": "enrich",
"input": { "connector": "c", "output": "data.x" }
}))
.expect_err("enrich has no `output` field");
let msg = err.to_string();
assert!(
msg.starts_with("config for function 'enrich':"),
"error should carry the function envelope, got: {msg}"
);
let ok = parse(json!({
"name": "enrich",
"input": { "connector": "c", "merge_path": "data.x" }
}))
.expect("merge_path is enrich's destination field");
assert!(matches!(ok, FunctionConfig::Enrich { .. }));
}
#[test]
fn publish_kafka_rejects_unknown_fields() {
let err = parse(json!({
"name": "publish_kafka",
"input": { "connector": "c", "topic": "t", "tpoic": "typo" }
}))
.expect_err("publish_kafka should reject an unknown field");
assert!(err.to_string().contains("unknown field"), "got: {err}");
}
#[test]
fn connector_is_returned_for_the_three_typed_integrations() {
let cases = [
(
json!({ "name": "http_call", "input": { "connector": "user_service" } }),
"user_service",
),
(
json!({ "name": "enrich",
"input": { "connector": "ref_data", "merge_path": "data.out" } }),
"ref_data",
),
(
json!({ "name": "publish_kafka",
"input": { "connector": "events", "topic": "t" } }),
"events",
),
];
for (input, expected) in cases {
let cfg = parse(input.clone()).expect("should parse");
assert_eq!(
cfg.connector().and_then(|c| c.as_static()),
Some(expected),
"for {input}"
);
}
}
#[test]
fn connector_is_none_for_every_non_connector_builtin() {
for name in BUILTIN_FUNCTION_NAMES {
if matches!(*name, "http_call" | "enrich" | "publish_kafka") {
continue;
}
let cfg = parse(json!({ "name": name, "input": minimal_input(name) }))
.unwrap_or_else(|e| panic!("'{name}' should parse: {e}"));
assert!(cfg.connector().is_none(), "'{name}' names no connector");
}
}
#[test]
fn connector_reads_the_custom_convention() {
let cfg = parse(json!({
"name": "pg_query",
"input": { "connector": "pg_main", "database": "orders" }
}))
.unwrap();
assert_eq!(cfg.connector().and_then(|c| c.as_static()), Some("pg_main"));
}
#[test]
fn connector_is_none_for_a_custom_input_without_a_string_connector() {
for input in [
json!({}), json!({ "connector": 7 }), json!({ "connector": true }), json!({ "connector": null }), json!({ "connector": ["a"] }), json!({ "connector": { "n": "a" } }), json!([]), json!(7), ] {
let cfg = parse(json!({ "name": "my_handler", "input": input.clone() }))
.unwrap_or_else(|e| panic!("custom should parse {input}: {e}"));
assert_eq!(cfg.connector(), None, "for input {input}");
}
}
#[test]
fn connector_returns_an_empty_name_verbatim() {
let typed = parse(json!({ "name": "http_call", "input": { "connector": "" } })).unwrap();
assert_eq!(typed.connector().and_then(|c| c.as_static()), Some(""));
let custom = parse(json!({ "name": "x", "input": { "connector": "" } })).unwrap();
assert_eq!(custom.connector().and_then(|c| c.as_static()), Some(""));
}
#[test]
fn connector_returns_a_non_ascii_name_byte_for_byte() {
let cfg =
parse(json!({ "name": "http_call", "input": { "connector": "連携先" } })).unwrap();
assert_eq!(cfg.connector().and_then(|c| c.as_static()), Some("連携先"));
}
#[test]
fn builtin_function_kind_is_none_for_non_builtins() {
for name in [
"",
"__not_a_builtin__",
"HTTP_CALL", "htttp_call", "map ", "publish_kafk", ] {
assert_eq!(
builtin_function_kind(name),
None,
"'{name}' must not classify as a built-in"
);
assert!(!is_builtin_function(name));
}
}
#[test]
fn builtin_kinds_partition_matches_the_sync_builtin_classifier() {
for name in BUILTIN_FUNCTION_NAMES {
let kind = builtin_function_kind(name)
.unwrap_or_else(|| panic!("'{name}' must classify as a built-in"));
let cfg = parse(json!({ "name": name, "input": minimal_input(name) }))
.unwrap_or_else(|e| panic!("'{name}' should parse with minimal input: {e}"));
assert_eq!(
cfg.is_sync_builtin(),
matches!(kind, BuiltinKind::SelfContained),
"'{name}' classifies as {kind:?} but is_sync_builtin() is {}",
cfg.is_sync_builtin()
);
}
}
#[test]
fn is_sync_builtin_agrees_with_arena_dispatch_for_every_builtin() {
use crate::engine::compiler::LogicCompiler;
use crate::engine::executor::with_arena;
use crate::engine::workflow::Workflow;
let names: Vec<&str> = BUILTIN_FUNCTION_NAMES
.iter()
.copied()
.chain(std::iter::once("some_custom_handler"))
.collect();
for name in names {
let workflow = Workflow::from_json(&format!(
r#"{{"id": "w", "name": "w", "priority": 0, "tasks": [
{{"id": "t", "name": "t", "function": {{"name": "{name}", "input": {}}}}}
]}}"#,
minimal_input(name)
))
.unwrap_or_else(|e| panic!("'{name}' should parse into a workflow: {e}"));
let compiler = LogicCompiler::new();
let compiled = compiler
.compile_workflows(vec![workflow])
.unwrap_or_else(|e| panic!("'{name}' should compile: {e}"));
let engine = compiler.into_engine();
let function = &compiled[0].tasks[0].function;
let mut message = Message::from_value(&json!({}));
let dispatches = with_arena(|arena| {
let mut arena_ctx = ArenaContext::from_owned(&message.context, arena);
function
.try_execute_in_arena(&mut message, &mut arena_ctx, &engine, None)
.is_some()
});
assert_eq!(
dispatches,
function.is_sync_builtin(),
"'{name}': is_sync_builtin() is {} but try_execute_in_arena() \
{} — the sync stretch would hit the engine-bug arm",
function.is_sync_builtin(),
if dispatches { "dispatched" } else { "declined" }
);
}
}
#[test]
fn requires_handler_kind_covers_exactly_the_config_only_integrations() {
for name in ["http_call", "enrich", "publish_kafka"] {
assert_eq!(
builtin_function_kind(name),
Some(BuiltinKind::RequiresHandler),
"'{name}' ships as config only and needs a registered handler"
);
}
for name in [
"map",
"validation",
"validate",
"parse_json",
"parse_xml",
"publish_json",
"publish_xml",
"filter",
"log",
] {
assert_eq!(
builtin_function_kind(name),
Some(BuiltinKind::SelfContained),
"'{name}' is executed by this crate"
);
}
}
}
#[cfg(test)]
mod dispatch_vocabulary_tests {
use super::*;
use std::collections::HashMap;
fn registry(names: &[&str]) -> HashMap<String, ()> {
names.iter().map(|n| ((*n).to_string(), ())).collect()
}
fn names(registry: &HashMap<String, ()>) -> Vec<&str> {
let mut out: Vec<&str> = dispatchable_functions_in(registry)
.map(|f| f.name)
.collect();
out.sort_unstable();
out
}
#[test]
fn aliases_and_canonical_names_agree() {
for name in BUILTIN_FUNCTION_NAMES {
let canonical = canonical_builtin_name(name);
assert_eq!(
canonical_builtin_name(canonical),
canonical,
"'{name}' resolves to '{canonical}', which must itself be canonical"
);
assert!(
BUILTIN_FUNCTION_NAMES.contains(&canonical),
"'{canonical}' is a canonical name and must be an accepted spelling"
);
assert_eq!(
builtin_function_kind(name),
builtin_function_kind(canonical),
"'{name}' and '{canonical}' are one function and must classify alike"
);
}
for name in BUILTIN_FUNCTION_NAMES {
let is_canonical = canonical_builtin_name(name) == *name;
let alias_of: Vec<&str> = BUILTIN_FUNCTION_NAMES
.iter()
.copied()
.filter(|c| builtin_aliases(c).contains(name))
.collect();
assert_eq!(
is_canonical,
alias_of.is_empty(),
"'{name}' must be canonical XOR an alias, got canonical={is_canonical} \
listed-as-alias-of={alias_of:?}"
);
assert!(
alias_of.len() <= 1,
"'{name}' is listed as an alias of more than one function: {alias_of:?}"
);
}
for canonical in BUILTIN_FUNCTION_NAMES {
for alias in builtin_aliases(canonical) {
assert_eq!(
canonical_builtin_name(alias),
*canonical,
"'{alias}' is listed under '{canonical}' but does not resolve to it"
);
}
}
}
#[test]
fn validate_is_canonical_and_validation_is_its_alias() {
assert_eq!(canonical_builtin_name("validation"), "validate");
assert_eq!(canonical_builtin_name("validate"), "validate");
assert_eq!(builtin_aliases("validate"), &["validation"]);
assert!(builtin_aliases("validation").is_empty());
assert!(builtin_aliases("map").is_empty());
}
#[test]
fn an_empty_registry_dispatches_every_self_contained_builtin() {
assert_eq!(
names(®istry(&[])),
vec![
"filter",
"log",
"map",
"parse_json",
"parse_xml",
"publish_json",
"publish_xml",
"validate",
],
"self-contained built-ins need no registration; `validation` is \
folded into `validate`, and the three config-only integrations are absent"
);
}
#[test]
fn requires_handler_builtins_appear_only_when_registered() {
let empty = registry(&[]);
assert!(!names(&empty).contains(&"enrich"));
assert!(!can_dispatch_in(&empty, "enrich"));
let backed = registry(&["enrich"]);
assert!(names(&backed).contains(&"enrich"));
assert!(can_dispatch_in(&backed, "enrich"));
let entry = dispatchable_functions_in(&backed)
.find(|f| f.name == "enrich")
.expect("registered enrich is enumerated");
assert_eq!(entry.kind, Some(BuiltinKind::RequiresHandler));
}
#[test]
fn custom_names_are_enumerated_with_no_kind() {
let reg = registry(&["shout"]);
let entry = dispatchable_functions_in(®)
.find(|f| f.name == "shout")
.expect("a registered custom name is enumerated");
assert_eq!(entry.kind, None, "None is how a custom handler reports");
assert!(entry.aliases.is_empty());
assert!(can_dispatch_in(®, "shout"));
assert!(!can_dispatch_in(®istry(&[]), "shout"));
}
#[test]
fn registering_a_self_contained_name_is_inert_and_never_duplicates_it() {
let shadowed = registry(&["map"]);
assert_eq!(
names(&shadowed),
names(®istry(&[])),
"a shadowing registration changes nothing about the vocabulary"
);
assert_eq!(
dispatchable_functions_in(&shadowed)
.filter(|f| f.name == "map")
.count(),
1,
"`map` is yielded exactly once, not once per source"
);
}
#[test]
fn aliases_dispatch_but_are_not_enumerated() {
let reg = registry(&[]);
assert!(
can_dispatch_in(®, "validation"),
"a task named `validation` really does execute"
);
assert!(
!names(®).contains(&"validation"),
"but the enumeration reports it under `validate`"
);
}
#[test]
fn can_dispatch_rejects_names_the_crate_does_not_know() {
let reg = registry(&["shout"]);
assert!(!can_dispatch_in(®, "SHOUT"), "matching is exact");
assert!(!can_dispatch_in(®, "htttp_call"));
assert!(!can_dispatch_in(®, ""));
}
#[test]
fn every_enumerated_name_is_dispatchable() {
let reg = registry(&["enrich", "shout"]);
for f in dispatchable_functions_in(®) {
assert!(
can_dispatch_in(®, f.name),
"'{}' is enumerated, so it must dispatch",
f.name
);
for alias in f.aliases {
assert!(
can_dispatch_in(®, alias),
"alias '{alias}' of '{}' must dispatch too",
f.name
);
}
}
}
}