82 lines
3.1 KiB
Rust
82 lines
3.1 KiB
Rust
//! 节点注册表 — 根据 node_type 字符串创建节点实例
|
||
|
||
use std::collections::HashMap;
|
||
|
||
use crate::dag::Dag;
|
||
use crate::dag_def::DagDef;
|
||
use crate::node::Node;
|
||
|
||
/// 节点工厂函数类型
|
||
type NodeFactory = Box<dyn Fn(&serde_json::Value) -> Box<dyn Node> + Send + Sync>;
|
||
|
||
/// 节点注册表 — 根据 node_type 字符串创建节点实例
|
||
pub struct NodeRegistry {
|
||
factories: HashMap<String, NodeFactory>,
|
||
}
|
||
|
||
impl NodeRegistry {
|
||
/// 创建空的注册表
|
||
pub fn new() -> Self {
|
||
Self {
|
||
factories: HashMap::new(),
|
||
}
|
||
}
|
||
|
||
/// 注册一个节点工厂
|
||
pub fn register<F>(&mut self, type_name: &str, factory: F)
|
||
where
|
||
F: Fn(&serde_json::Value) -> Box<dyn Node> + Send + Sync + 'static,
|
||
{
|
||
self.factories.insert(type_name.to_string(), Box::new(factory));
|
||
}
|
||
|
||
/// 根据 node_type 创建节点实例
|
||
pub fn create(&self, type_name: &str, config: &serde_json::Value) -> anyhow::Result<Box<dyn Node>> {
|
||
self.factories
|
||
.get(type_name)
|
||
.ok_or_else(|| anyhow::anyhow!("未注册的节点类型: {}", type_name))
|
||
.map(|factory| factory(config))
|
||
}
|
||
|
||
/// 从 DagDef 构建完整的运行时 Dag
|
||
pub fn build_dag(&self, def: &DagDef) -> anyhow::Result<Dag> {
|
||
let mut dag = Dag::new();
|
||
|
||
// 创建所有节点实例，同时收集节点 ID 集合用于边校验
|
||
let mut node_ids: std::collections::HashSet<&String> = std::collections::HashSet::new();
|
||
for (id, node_def) in &def.nodes {
|
||
let node = self.create(&node_def.node_type, &node_def.config)?;
|
||
// 节点级 config 下沉到 Dag.node_configs：DagExecutor.run 构造 NodeContext 时
|
||
// deep_merge(initial_config, node_configs[id])（节点级覆盖全局级），解决此前
|
||
// initial_config.clone() 直接当 ctx.config、忽略 NodeDef.config 的问题。
|
||
dag.add_node_with_config(id.clone(), node, node_def.config.clone());
|
||
node_ids.insert(id);
|
||
}
|
||
|
||
// 添加所有边 — 校验两端节点均存在，野边早返回
|
||
for edge_def in &def.edges {
|
||
if !node_ids.contains(&edge_def.source) || !node_ids.contains(&edge_def.target) {
|
||
anyhow::bail!(
|
||
"边 source/target 节点不存在: {} -> {}",
|
||
edge_def.source,
|
||
edge_def.target
|
||
);
|
||
}
|
||
match &edge_def.condition {
|
||
Some(cond) => dag.add_edge_with_condition(
|
||
edge_def.source.clone(),
|
||
edge_def.target.clone(),
|
||
cond.clone(),
|
||
),
|
||
None => dag.add_edge(edge_def.source.clone(), edge_def.target.clone()),
|
||
}
|
||
}
|
||
|
||
Ok(dag)
|
||
}
|
||
}
|
||
|
||
// 不实现 Default:原 Default 注册了一个会 panic 的 script 工厂(unimplemented!),
|
||
// 违反项目铁律「无 panic——所有占位代码返回空/默认值」。所有调用方应显式 new() +
|
||
// 手动 register 真实节点(如 state.rs::build_registry 的做法)。
|