feat: 完善RPA编排与本地能力接入
This commit is contained in:
1 parent
6a2cc24898
commit
c03380168d
32 files changed
+2763
-185
No files matched your search
+36
-26
@@ -137,12 +137,8 @@ export async function runFlow(
|
||||
|
||||
options.emit({ type: "run.started", startedAt });
|
||||
|
||||
const order = topLevelOrder(flow);
|
||||
try {
|
||||
for (const node of order) {
|
||||
if (options.signal.aborted) throw new RuntimeError("cancelled", "运行已取消");
|
||||
await executeNode(node, ctx, states);
|
||||
}
|
||||
await executeTopLevelGraph(flow, ctx, states);
|
||||
} catch (error) {
|
||||
if (error instanceof StopSignal) {
|
||||
options.log("流程被「退出流程」节点终止", "warn");
|
||||
@@ -179,33 +175,47 @@ export async function runFlow(
|
||||
};
|
||||
}
|
||||
|
||||
function topLevelOrder(flow: FlowDefinition): FlowNode[] {
|
||||
async function executeTopLevelGraph(flow: FlowDefinition, ctx: RunContext, states: Map<string, NodeState>): Promise<void> {
|
||||
const top = flow.nodes.filter((node) => !node.parentId);
|
||||
if (top.length <= 1) return top;
|
||||
const topIds = new Set(top.map((node) => node.id));
|
||||
const incoming = new Map<string, number>();
|
||||
for (const edge of flow.edges) {
|
||||
if (topIds.has(edge.source) && topIds.has(edge.target)) {
|
||||
incoming.set(edge.target, (incoming.get(edge.target) || 0) + 1);
|
||||
if (!top.length) return;
|
||||
if (!flow.edges.length) {
|
||||
for (const node of [...top].sort((a, b) => a.position.y - b.position.y || a.position.x - b.position.x)) {
|
||||
if (ctx.signal.aborted) throw new RuntimeError("cancelled", "运行已取消");
|
||||
await executeNode(node, ctx, states);
|
||||
}
|
||||
return;
|
||||
}
|
||||
const topIds = new Set(top.map((node) => node.id));
|
||||
const topById = new Map(top.map((node) => [node.id, node]));
|
||||
const edges = flow.edges.filter((edge) => topIds.has(edge.source) && topIds.has(edge.target));
|
||||
const incoming = new Set(edges.map((edge) => edge.target));
|
||||
const start =
|
||||
top.find((node) => !incoming.get(node.id) && node.componentId.startsWith("Trigger.")) ||
|
||||
top.find((node) => !incoming.get(node.id)) ||
|
||||
top.find((node) => !incoming.has(node.id) && node.componentId.startsWith("Trigger.")) ||
|
||||
top.find((node) => !incoming.has(node.id)) ||
|
||||
top[0];
|
||||
const ordered: FlowNode[] = [];
|
||||
const visited = new Set<string>();
|
||||
let current: FlowNode | undefined = start;
|
||||
while (current && !visited.has(current.id)) {
|
||||
ordered.push(current);
|
||||
visited.add(current.id);
|
||||
const edge = flow.edges.find((item) => item.source === current!.id && topIds.has(item.target));
|
||||
current = edge ? top.find((node) => node.id === edge.target) : undefined;
|
||||
}
|
||||
for (const node of [...top].sort((a, b) => a.position.y - b.position.y)) {
|
||||
if (!visited.has(node.id)) ordered.push(node);
|
||||
}
|
||||
return ordered;
|
||||
|
||||
const walk = async (node: FlowNode): Promise<void> => {
|
||||
if (visited.has(node.id)) return;
|
||||
if (ctx.signal.aborted) throw new RuntimeError("cancelled", "运行已取消");
|
||||
visited.add(node.id);
|
||||
try {
|
||||
await executeNode(node, ctx, states);
|
||||
} catch (error) {
|
||||
const errorEdge = edges.find((edge) => edge.source === node.id && edge.sourceHandle === "error");
|
||||
const target = errorEdge ? topById.get(errorEdge.target) : undefined;
|
||||
if (!target) throw error;
|
||||
ctx.vars.error = asRuntimeError(error, node, node.componentId);
|
||||
ctx.log(`节点失败,转入异常分支:${target.label || target.componentId}`, "warn", node.id);
|
||||
await walk(target);
|
||||
return;
|
||||
}
|
||||
const normalEdge = edges.find((edge) => edge.source === node.id && edge.sourceHandle !== "error");
|
||||
const target = normalEdge ? topById.get(normalEdge.target) : undefined;
|
||||
if (target) await walk(target);
|
||||
};
|
||||
|
||||
await walk(start);
|
||||
}
|
||||
|
||||
async function executeNode(node: FlowNode, ctx: RunContext, states: Map<string, NodeState>): Promise<unknown> {
|
||||
|
||||
Reference in new issue
Block a user