reifydb_sub_flow/
builder.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{collections::HashMap, path::PathBuf, sync::Arc};
5
6use reifydb_core::interface::catalog::flow::FlowNodeId;
7use reifydb_sdk::{
8	config::Config,
9	connector::{
10		sink::{FFISink, FFISinkMetadata},
11		source::{FFISource, FFISourceMetadata},
12	},
13};
14use reifydb_value::Result;
15
16use crate::{connector::ConnectorRegistry, operator::BoxedOperator};
17
18pub type OperatorFactory = Arc<dyn Fn(FlowNodeId, &Config) -> Result<BoxedOperator> + Send + Sync>;
19
20#[derive(Clone)]
21pub struct CustomOperators {
22	inner: Arc<HashMap<String, OperatorFactory>>,
23}
24
25impl CustomOperators {
26	pub fn new(map: HashMap<String, OperatorFactory>) -> Self {
27		Self {
28			inner: Arc::new(map),
29		}
30	}
31
32	pub fn get(&self, name: &str) -> Option<&OperatorFactory> {
33		self.inner.get(name)
34	}
35}
36
37pub struct FlowConfigurator {
38	operators_dir: Option<PathBuf>,
39	custom_operators: HashMap<String, OperatorFactory>,
40	connector_registry: ConnectorRegistry,
41}
42
43impl Default for FlowConfigurator {
44	fn default() -> Self {
45		Self::new()
46	}
47}
48
49impl FlowConfigurator {
50	pub fn new() -> Self {
51		Self {
52			operators_dir: None,
53			custom_operators: HashMap::new(),
54			connector_registry: ConnectorRegistry::new(),
55		}
56	}
57
58	pub fn operators_dir(mut self, path: PathBuf) -> Self {
59		self.operators_dir = Some(path);
60		self
61	}
62
63	pub fn register_operator(
64		mut self,
65		name: impl Into<String>,
66		factory: impl Fn(FlowNodeId, &Config) -> Result<BoxedOperator> + Send + Sync + 'static,
67	) -> Self {
68		self.custom_operators.insert(name.into(), Arc::new(factory));
69		self
70	}
71
72	pub fn register_source<S: FFISource + FFISourceMetadata>(mut self) -> Self {
73		self.connector_registry.register_source::<S>();
74		self
75	}
76
77	pub fn register_sink<S: FFISink + FFISinkMetadata>(mut self) -> Self {
78		self.connector_registry.register_sink::<S>();
79		self
80	}
81
82	pub(crate) fn configure(self) -> FlowConfig {
83		FlowConfig {
84			operators_dir: self.operators_dir,
85			custom_operators: self.custom_operators,
86			connector_registry: self.connector_registry,
87		}
88	}
89}
90
91pub struct FlowConfig {
92	pub operators_dir: Option<PathBuf>,
93
94	pub custom_operators: HashMap<String, OperatorFactory>,
95
96	pub connector_registry: ConnectorRegistry,
97}
reifydb_sub_flow/builder.rs

reifydb_sub_flow/
builder.rs
