reifydb_sub_flow/operator/
provider.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4#[cfg(all(reifydb_target = "host", not(reifydb_dst)))]
5use std::sync::Arc;
6
7#[cfg(all(reifydb_target = "host", not(reifydb_dst)))]
8use reifydb_codec::value::encode_params;
9use reifydb_core::interface::catalog::flow::OperatorId;
10#[cfg(all(reifydb_target = "host", not(reifydb_dst)))]
11use reifydb_extension::operator::extern_c::loader::extern_c_operator_loader;
12#[cfg(all(reifydb_target = "host", not(reifydb_dst)))]
13use reifydb_flow::error::FlowStateError;
14use reifydb_flow::{
15	error::FlowGraphError,
16	operator::{BoxedHostOperator, provider::OperatorProvider},
17};
18#[cfg(all(reifydb_target = "host", not(reifydb_dst)))]
19use reifydb_value::params::Params;
20use reifydb_value::{Result, config::Config, error::Error};
21#[cfg(all(reifydb_target = "host", not(reifydb_dst)))]
22use tracing::instrument;
23
24use crate::builder::CustomOperators;
25#[cfg(all(reifydb_target = "host", not(reifydb_dst)))]
26use crate::{
27	error::ExternOperatorError, operator::extern_c::ExternCOperatorHandle,
28	operator::extern_rust::extern_rust_operator_loader,
29};
30
31pub struct StandardOperatorProvider {
32	custom: CustomOperators,
33}
34
35impl StandardOperatorProvider {
36	pub fn new(custom: CustomOperators) -> Self {
37		Self {
38			custom,
39		}
40	}
41}
42
43impl OperatorProvider for StandardOperatorProvider {
44	fn provide(&self, operator_id: OperatorId, config: &Config) -> Result<BoxedHostOperator> {
45		let operator = config.name();
46
47		if let Some(factory) = self.custom.get(operator) {
48			return factory(operator_id, config);
49		}
50
51		#[cfg(all(reifydb_target = "host", not(reifydb_dst)))]
52		{
53			if extern_rust_operator_loader().read().has_operator(operator) {
54				return extern_rust_operator_loader().write().create_operator_by_name(
55					operator,
56					operator_id,
57					config,
58				);
59			}
60
61			if extern_c_operator_loader().read().has_operator(operator) {
62				return self.create_extern_c_operator(operator, operator_id, config);
63			}
64
65			Err(Error::from(FlowGraphError::UnknownOperator {
66				operator: operator.to_string(),
67			}))
68		}
69
70		#[cfg(not(all(reifydb_target = "host", not(reifydb_dst))))]
71		{
72			Err(Error::from(FlowGraphError::ExternUnsupportedOnWasm))
73		}
74	}
75}
76
77#[cfg(all(reifydb_target = "host", not(reifydb_dst)))]
78impl StandardOperatorProvider {
79	#[instrument(name = "flow::provider::create_extern_c_operator", level = "debug", skip(self, config), fields(operator = %operator, operator_id = ?operator_id))]
80	fn create_extern_c_operator(
81		&self,
82		operator: &str,
83		operator_id: OperatorId,
84		config: &Config,
85	) -> Result<BoxedHostOperator> {
86		let loader = extern_c_operator_loader();
87		let mut loader_write = loader.write();
88
89		let config_params =
90			Params::Named(Arc::new(config.iter().map(|(k, v)| (k.clone(), v.clone())).collect()));
91		let config_bytes = encode_params(&config_params).map_err(|e| {
92			Error::from(FlowStateError::Encode {
93				state: "operator config",
94				cause: e.to_string(),
95			})
96		})?;
97
98		let created = loader_write.create_operator_by_name(operator, operator_id, &config_bytes);
99		let (descriptor, instance) = match created {
100			Ok(created) => created,
101			Err(e) => {
102				return Err(Error::from(ExternOperatorError::CreateFailed {
103					cause: format!("{:?}", e),
104				}));
105			}
106		};
107
108		Ok(Box::new(ExternCOperatorHandle::new(descriptor, instance, operator_id)))
109	}
110}
reifydb_sub_flow/operator/provider.rs

reifydb_sub_flow/operator/
provider.rs
