前、后端通用的可视化逻辑编排

前一段时间写过一篇文章《实战，一个高扩展、可视化低代码前端，详实、完整》，得到了很多朋友的关注。
其中的逻辑编排部分过于简略，不少朋友希望能写一些关于逻辑编排的内容，本文就详细讲述一下逻辑编排的实现原理。
逻辑编排的目的，是用最少甚至不用代码来实现软件的业务逻辑，包括前端业务逻辑跟后端业务逻辑。本文前端代码基于typescript、react技术栈，后端基于golang。
涵盖内容：数据流驱动的逻辑编排原理，业务编排编辑器的实现，页面控件联动，前端业务逻辑与UI层的分离，子编排的复用、自定义循环等嵌入式子编排的处理、事务处理等
运行快照：

前端项目地址：https://github.com/codebdy/rxdrag
前端演示地址：https://rxdrag.vercel.app/
后端演示尚未提供，代码地址：https://github.com/codebdy/minions-go
注：为了便于理解，本文使用的代码做了简化处理，会跟实际代码有些细节上的出入。
整体架构

整个逻辑编排，由以下几部分组成：

• 节点物料，用于定义编辑器中的元件，包含在工具箱中的图标，端口以及属性面板中的组件schema。
  
• 逻辑编排编辑器，顾名思义，可视化编辑器，根据物料提供的元件信息，编辑生成JSON格式的“编排描述数据”。
  
• 编排描述数据，用户操作编辑器的生成物，供解析引擎消费
  
• 前端解析引擎，Typescript 实现的解析引擎，直接解析“编排描述数据”并执行，从而实现的软件的业务逻辑。
  
• 后端解析引擎，Golang 实现的解析引擎，直接解析“编排描述数据”并执行，从而实现的软件的业务逻辑。
  

逻辑编排实现方式的选择

逻辑编排，实现方式很多，争议也很多。
一直以来，小编的思路也很局限。从流程图层面，以线性的思维去思考，认为逻辑编排的意义并不大。因为，经过这么多年发展，事实证明代码才是表达逻辑的最佳方式，没有之一。用流程图去表达代码，最终只能是老板、客户的丰满理想与程序员骨感现实的对决。
直到看到Mybricks项目交互部分的实现方式，才打开了思路。类似unreal蓝图数据流驱动的实现方式，其实大有可为。
这种方式的意义是，跳出循环、if等这些底层的代码细节，以数据流转的方式思考业务逻辑，从而把业务逻辑抽象为可复用的组件，每个组件对数据进行相应处理或者根据数据执行相应动作，从而达到复用业务逻辑的目的。并且，节点的粒度可大可小，非常灵活。
具体实现方式是，把每个逻辑组件看成一个黑盒，通过入端口流入数据，出端口流出变换后的数据：

举个例子，一个节点用来从数据库查询客户列表，会是这样的形式：

用户不需要关注这个元件节点的实现细节，只需要知道每个端口的功能就可以使用。这个元件节点的功能可以做的很简单，比如一个fetch，只有几十行代码。也可以做到很强大，比如类似useSwr，自带缓存跟状态管理，可以有几百甚至几千行代码。
我们希望这些元件节点是可以自行定义，方便插入的，并且我们做到了。
出端口跟入端口之间，可以用线连接，表示元件节点之间的调用关系，或者说是数据的流入关系。假如，数据读取成功，需要显示在列表中；失败，提示错误消息；查询时，显示等待的Spinning，那么就可以再加三个元件节点，变成：

如果用流程图，上面这个编排，会被显示成如下样子：

两个比较，就会发现，流程图的思考方式，会把人引入条件细节，其实就是试图用不擅长代码的图形来描述代码。是纯线性的，没有回调，也就无法实现类似js promise的异步。
而数据流驱动的逻辑编排，可以把人从细节中解放出来，用模块化的思考方式去设计业务逻辑，更方便把业务逻辑拆成一个个可复用的单元。
如果以程序员的角度来比喻，流程图相当于一段代码脚本，是面向过程的；数据流驱动的逻辑编排像是几个类交互完成一个功能，更有点面向对象的感觉。
朋友，如果是让你选，你喜欢哪种方式？欢迎留言讨论。
另外还有一种类似stratch的实现方式：

感觉这种纯粹为了可视化而可视化，只适合小孩子做玩具。会写代码的人不愿意用，太低效了。不会写代码的人，需要理解代码才会用。适合场景是用直观的方式介绍什么是代码逻辑，就是说只适合相对比较低智力水平的编程教学，比如幼儿园、小学等。商业应用，就免了。
数据流驱动的逻辑编排

一个简单的例子

从现在开始，放下流程图，忘记strach，我们从业务角度去思考也逻辑，然后设计元件节点去实现相应的逻辑。
选一个简单又典型的例子：学生成绩单。一个成绩单包含如下数据：

假如数据已经从数据库取出来了，第一步处理，统计每个学生的总分数。设计这么几个元件节点来配合完成：

这个编排，输入成绩列表，循环输出每个学生的总成绩。为了完成这个编排，设计了四个元件节点：

• 循环，入端口接收一个列表，遍历列表并循环输出，每一次遍历往“单次输出”端口发送一条数据，可以理解为一个学生对象（尽量从对象的角度思考，而不是数据记录），遍历结束后往“结束端口”发送循环的总数。如果按照上面的列表，“单次输出端口”会被调用4次，每次输出一个学生对象{姓名：xxx，语文：xxx，数学：xxx...}，“结束”端口只被调用一次，输出结果是 4.
  
• 拆分对象，这个元件节点的出端口是可以动态配置的，它的功能是把一个对象按照属性值按照名字分发到指定的出端口。本例中，就是把各科成绩拆分开来。
  
• 收集数组，这个节点也可以叫收集到数组，作用是把串行接收到的数据组合到一个数组里。他有两个入端口：input端口，用来接收串行输入，并缓存到数组；finished端口，表示输入完成，把缓存到的数据组发送给输出端口。
  
• 加和，把输入端口传来的数组进行加和计算，输出总数。
  

这是一种跟代码完全不同的思考方式，每一个元件节点，就是一小段业务逻辑，也就是所谓的业务逻辑组件化。我们的项目中，只提供给了有限的预定义元件节点，想要更多的节点，可以自行自定义并注入系统，具体设计什么样的节点，完全取决于用户的业务需求跟喜好。作者更希望设计元件的过程是一个创作的过程，或许具备一定的艺术性。
刚刚的例子，审视之。有人可能会换一个方式来实现，比如拆分对象跟收集数据这两个节点，合并成一个节点：对象转数组，可能更方便，适应能力也更强：

对象转换数组节点，对象属性与数组索引的对应关系，可以通过属性面板的配置来完成。
这两种实现方式，说不清哪种更好，选择自己喜欢的，或者两种都提供。
输入节点、输出节点

一段图形化的逻辑编排，通过解析引擎，会被转换成一段可执行的业务逻辑。这段业务逻辑需要跟外部对接，为了明确对接语义，再添加两个特殊的节点元件：输入节点（开始节点），输出节点（结束节点）。

输入节点用于标识逻辑编排的入口，输入节点可以有一个或者多个，输入节点用细线圆圈表示。
输出节点用于标识逻辑编排的出口，输出节点可以有一个或者多个，输出节点用粗线圆圈表示。
在后面的引擎部分，会详细描述输入跟输出节点如何跟外部的对接。
编排的复用：子编排

一般低代码中，提升效率的方式是复用，尽可能复用已有的东西，比如组件、业务逻辑，从而达到降本、增效的目的。
设计元件节点是一种创作，那么使用元件节点进行业务编排，更是一种基于领域的创作。辛辛苦苦创作的编排，如果能被复用，应该算是对创作本身的尊重吧。
如果编排能够像元件节点一样，被其它逻辑编排所引用，那么这样的复用方式无疑是最融洽的。也是最方便的实现方式。
把能够被其它编排引用的编排称为子编排，上面计算学生总成绩的编排，转换成子编排，被引入时的形态应该是这样的：

子编排元件的输入端口对应逻辑编排实现的输入节点，输出端口对应编排实现的输出节点。
嵌入式编排节点

前文设计的循环组件非常简单，循环直接执行到底，不能被中断。但是，有的时候，在处理数据的时候，要根据每次遍历到的数据做判断，来决定继续循环还是终止循环。
就是说，需要一个循环节点，能够自定义它的处理流程。依据这个需求，设计了自定义循环元件，这是一种能够嵌入编排的节点，形式如下：

这种嵌入式编排节点，跟其它元件节点一样，事先定义好输入节点跟输出节点。只是它不完全是黑盒，其中一部分通过逻辑编排这种白盒方式来实现。
这种场景并不多见，除了循环，后端应用中，还有事务元件也需要类似实现方式：

嵌入式元件跟其它元件节点一样，可以被其它元件连接，嵌入式节点在整个编排中的表现形式：

基本概念

为了进一步深入逻辑编排引擎跟编辑器的实现原理，先梳理一些基本的名词、概念。
逻辑编排，本文特指数据流驱动的逻辑编排，是由图形表示的一段业务逻辑，由元件节点跟连线组成。
元件节点，简称元件、节点、编排元件、编排单元。逻辑编排中具体的业务逻辑处理单元，带副作用的，可以实现数据转换、页面组件操作、数据库数据存取等功能。一个节点包含零个或多个输入端口，包含零个或多个输出端口。在设计其中，以圆角方形表示：

端口，分为输入端口跟输出端口两种。是元件节点流入或流出数据的通道（或者接口）。在逻辑单元中，用小圆圈表示。
输入端口，简称入端口、入口。输入端口位于元件节点的左侧。
输出端口，简称出端口、出口。输出端口位于元件节点的右侧。
单入口元件，只有一个入端口的元件节点。
多入口元件，有多个入端口的元件节点。
单出口元件，只有一个出端口的元件节点。
多出口元件，有多个出端口的元件节点。
输入节点，一种特殊的元件节点，用于描述逻辑编排的起点（开始点）。转换成子编排后，会对应子编排相应的入端口。
输出节点，一种特殊的元件节点，用于描述逻辑编排的终点（结束点）。转换成子编排后，会对应子编排相应的出端口。
嵌入式编排，特殊的元件节点，内部实现由逻辑编排完成。示例：

子编排，特殊的逻辑编排，该编排可以转换成元件节点，供其它逻辑编排使用。
连接线，简称连线、线。用来连接各个元件节点，表示数据的流动关系。
定义DSL

逻辑编排编辑器生成一份JSON，解析引擎解析这份JSON，把图形化的业务逻辑转化成可执行的逻辑，并执行。
编辑器跟解析引擎之间要有份约束协议，用来约定JSON的定义，这个协议就是这里定义的DSL。在typescript中，用interface、enum等元素来表示。
这些DSL仅仅是用来描述页面上的图形元素，通过activityName属性跟具体的实现代码逻辑关联起来。比如一个循环节点，它的actvityName是Loop，解析引擎会根据Loop这个名字找到该节点对应的实现类，并实例化为一个可执行对象。后面的解析引擎会详细展开描述这部分。
节点类型

元件节点类型叫NodeType，用来区分不同类型的节点，在TypeScript中是一个枚举类型。

1. export enum NodeType {
2.   //开始节点
3.   Start = 'Start',
4.   //结束节点
5.   End = 'End',
6.   //普通节点
7.   Activity = 'Activity',
8.   //子编排，对其它编排的引用
9.   LogicFlowActivity = "LogicFlowActivity",
10.   //嵌入式节点，比如自定义逻辑编排
11.   EmbeddedFlow = "EmbeddedFlow"
12. }

复制代码

端口

1. export interface IPortDefine {
2.   //唯一标识
3.   id: string;
4.   //端口名词
5.   name: string;
6.   //显示文本
7.   label?: string;
8. }

复制代码

元件节点

1. //一段逻辑编排数据
2. export interface ILogicFlowMetas {
3.   //所有节点
4.   nodes: INodeDefine<unknown>[];
5.   //所有连线
6.   lines: ILineDefine[];
7. }
8. export interface INodeDefine<ConfigMeta = unknown> {
9.   //唯一标识
10.   id: string;
11.   //节点名称，一般用于开始结束、节点，转换后对应子编排的端口
12.   name?: string;
13.   //节点类型
14.   type: NodeType;
15.   //活动名称，解析引擎用，通过该名称，查找构造节点的具体运行实现
16.   activityName: string;
17.   //显示文本
18.   label?: string;
19.   //节点配置
20.   config?: ConfigMeta;
21.   //输入端口
22.   inPorts?: IPortDefine[];
23.   //输出端口
24.   outPorts?: IPortDefine[];
25.   //父节点，嵌入子编排用
26.   parentId?: string;
27.   // 子节点，嵌入编排用
28.   children?: ILogicFlowMetas
29. }

复制代码

连接线

1. //连线接头
2. export interface IPortRefDefine {
3.   //节点Id
4.   nodeId: string;
5.   //端口Id
6.   portId?: string;
7. }
9. //连线定义
10. export interface ILineDefine {
11.   //唯一标识
12.   id: string;
13.   //起点
14.   source: IPortRefDefine;
15.   //终点
16.   target: IPortRefDefine;
17. }

复制代码

逻辑编排

1. //这个代码上面出现过，为了使extends更直观，再出现一次
2. //一段逻辑编排数据
3. export interface ILogicFlowMetas {
4.   //所有节点
5.   nodes: INodeDefine<unknown>[];
6.   //所有连线
7.   lines: ILineDefine[];
8. }
9. //逻辑编排
10. export interface ILogicFlowDefine extends ILogicFlowMetas {
11.   //唯一标识
12.   id: string;
13.   //名称
14.   name?: string;
15.   //显示文本
16.   label?: string;
17. }

复制代码

解析引擎的实现

解析引擎有两份实现：Typescript实现跟Golang实现。这里介绍基于原理，以Typescript实现为准，后面单独章节介绍Golang的实现方式。也有朋友根据这个dsl实现了C#版自用，欢迎朋友们实现不同的语言版本并开源。
DSL只是描述了节点跟节点之间的连接关系，业务逻辑的实现，一点都没有涉及。需要为每个元件节点制作一个单独的处理类，才能正常解析运行。比如上文中的循环节点，它的DSL应该是这样的：

1. {
2.   "id": "id-1",
3.   "type": "Activity",
4.   "activityName": "Loop",
5.   "label": "循环",
6.   "inPorts": [
7.     {
8.       "id":"port-id-1",
9.       "name":"input",
10.       "label":""
11.     }
12.   ],
13.   "outPorts": [
14.     {
15.       "id":"port-id-2",
16.       "name":"output",
17.       "label":"单次输出"
18.     },
19.     {
20.       "id":"port-id-3",
21.       "name":"finished",
22.       "label":"结束"
23.     }
24.   ]
25. }

复制代码

开发人员制作一个处理类LoopActivity用来处理循环节点的业务逻辑，并将这个类注册入解析引擎，key为loop。这个类，我们叫做活动（Activity）。解析引擎，根据activityName查找类，并创建实例。LoopActivity的类实现应该是这样：

1. export interface IActivity{
2.   inputHandler (inputValue?: unknown, portName:string);
3. }
4. export class LoopActivity implements IActivity{
5.   constructor(protected meta: INodeDefine<ILoopConfig>) {}
6.   //输入处理
7.   inputHandler (inputValue?: unknown, portName:string){
8.     if(portName !== "input"){
9.       console.error("输入端口名称不正确")
10.       return      
11.     }
12.     let count = 0
13.     if (!_.isArray(inputValue)) {
14.       console.error("循环的输入值不是数组")
15.     } else {
16.       for (const one of inputValue) {
17.         this.output(one)
18.         count++
19.       }
20.     }
21.     //输出循环次数
22.     this.next(count, "finished")
23.   }
24.   //单次输出
25.   output(value: unknown){
26.     this.next(value, "output")
27.   }
28.   
29.   next(value:unknown, portName:string){
30.      //把数据输出到指定端口,这里需要解析器注入代码
31.   }
32. }

复制代码

解析引擎根据DSL，调用inputHanlder，把控制权交给LoopActivity的对象，LoopActivity处理完成后把数据通过next方法传递出去。它只需要关注自身的业务逻辑就可以了。
这里难点是，引擎如何让所有类似LoopActivity类的对象联动起来。这个实现是逻辑编排的核心，虽然实现代码只有几百行，但是很绕，需要静下心来好好研读接下来的部分。
编排引擎的设计

编排引擎类图

LogicFlow类，代表一个完整的逻辑编排。它解析一张逻辑编排图，并执行该图所代表的逻辑。
IActivity接口，一个元件节点的执行逻辑。不同的逻辑节点，实现不同的Activity类，这类都实现IActivity接口。比如循环元件，可以实现为

1. export class LoopActivity implements IActivity{
2.     id: string
3.     config: LoopActivityConfig
4. }

复制代码

LogicFlow类解析逻辑编排图时，根据解析到的元件节点，创建相应的IActivity实例，比如解析到Loop节点的时候，就创建LoopActivity实例。
LogicFlow还有一个功能，就是根据连线，给构建的IActivity实例建立连接关系，让数据能在不同的IActivity实例之间流转。先明白引擎中的数据流，是理解上述类图的前提。
解析引擎中的Jointer

在解析引擎中，数据按照以下路径流动：

有三个节点：节点A、节点B、节点C。数据从节点A的“a-in-1”端口流入，通过一些处理后，从节点A的“a-out-1”端口流出。在“a-out-1”端口，把数据分发到节点B的“b-in-1”端口跟节点C的“c-in-1”端口。在B、C节点以后，继续重复类似的流动。
端口“a-out-1”要把数据分发到端口“b-in-1”和端口“c-in-1”，那么端口“a-out-1”要保存端口“b-in-1”和端口“c-in-1”的引用。就是说在解析引擎中，端口要建模为一个类，端口“a-out-1”是这个类的对象。要想分发数据，端口类跟自身是一个聚合关系。这种关系，让解析引擎中的端口看起来像连接器，故取名Jointer。一个Joniter实例，对应一个元件节点的端口。
在逻辑编排图中，一个端口，可以连接多个其它端口。所以，一个Jointer也可以连接多个其它Jointer。

注意，这是实例的关系，如果对应到类图，就是这样的关系：

Jointer通过调用push方法把数据传递给其他Jointer实例。
connect方法用于给两个Joiner构建连接关系。
用TypeScript实现的话，代码是这样的：

1. //数据推送接口
2. export type InputHandler = (inputValue: unknown, context?:unknown) => void;
3. export interface IJointer {
4.   name: string;
5.   //接收上一级Jointer推送来的数据
6.   push: InputHandler;
7.   //添加下游Jointer
8.   connect: (jointerInput: InputHandler) => void;
9. }
11. export class Jointer implements IJointer {
12.   //下游Jonter的数据接收函数
13.   private outlets: IJointer[] = []
15.   constructor(public id: string, public name: string) {
16.   }
18.   //接收上游数据，并分发到下游
19.   push: InputHandler = (inputValue?: unknown, context?:unknown) => {
20.     for (const jointer of this.outlets) {
21.       //推送数据
22.       jointer.push(inputValue, context)
23.     }
24.   }
26.   //添加下游Joninter
27.   connect = (jointer: IJointer) => {
28.     //往数组加数据，跟上面的push不一样
29.     this.outlets.push(jointer)
30.   }
32.   //删除下游Jointer
33.   disconnect = (jointer: InputHandler) => {
34.     this.outlets.splice(this.outlets.indexOf(jointer), 1)
35.   }
36. }

复制代码

在TypeScript跟Golang中，函数是一等公民。但是在类图里面，这个独立的一等公民是不好表述的。所以，上面的代码只是对类图的简单翻译。在实现时，Jointer的outlets可以不存IJointer的实例，只存Jointer的push方法，这样的实现更灵活，并且更容易把一个逻辑编排转成一个元件节点，优化后的代码：

1. //数据推送接口
2. export type InputHandler = (inputValue: unknown, context?:unknown) => void;
4. export interface IJointer {
5.   //当key使用，不参与业务逻辑
6.   id: string;
7.   name: string;
8.   //接收上一级Jointer推送来的数据
9.   push: InputHandler;
10.   //添加下游Jointer
11.   connect: (jointerInput: InputHandler) => void;
12. }
14. export class Jointer implements IJointer {
15.   //下游Jonter的数据接收函数
16.   private outlets: InputHandler[] = []
18.   constructor(public id: string, public name: string) {
19.   }
21.   //接收上游数据，并分发到下游
22.   push: InputHandler = (inputValue?: unknown, context?:unknown) => {
23.     for (const jointerInput of this.outlets) {
24.       jointerInput(inputValue, context)
25.     }
26.   }
28.   //添加下游Joninter
29.   connect = (inputHandler: InputHandler) => {
30.     this.outlets.push(inputHandler)
31.   }
33.   //删除下游Jointer
34.   disconnect = (jointer: InputHandler) => {
35.     this.outlets.splice(this.outlets.indexOf(jointer), 1)
36.   }
37. }

复制代码

记住这里的优化：Jointer的下游已经不是Jointer了，是Jointer的push方法，也可以是独立的其它方法，只要参数跟返回值跟Jointer的push方法一样就行，都是InputHandler类型。这个优化，可以让把Activer的某个处理函数设置为入Jointer的下游，后面会有进一步介绍。
Activity与Jointer的关系

一个元件节点包含多个（或零个）入端口和多个（或零个）出端口。那么意味着一个IActivity实例包含多个Jointer，这些Jointer也按照输入跟输出来分组：

TypeScript定义的代码如下：

1. export interface IActivityJointers {
2.   //入端口对应的连接器
3.   inputs: IJointer[];
4.   //处端口对应的连接器
5.   outputs: IJointer[];
7.   //通过端口名获取出连接器
8.   getOutput(name: string): IJointer | undefined
9.   //通过端口名获取入连接器
10.   getInput(name: string): IJointer | undefined
11. }
13. //活动接口，一个实例对应编排图一个元件节点，用于实现元件节点的业务逻辑
14. export interface IActivity<ConfigMeta = unknown> {
15.   id: string;
16.   //连接器，跟元件节点的端口异议对应
17.   jointers: IActivityJointers,
18.   //元件节点配置，每个Activity的配置都不一样，故而用泛型
19.   config?: ConfigMeta;
20.   //销毁
21.   destory(): void;
22. }

复制代码

入端口挂接业务逻辑

入端口对应一个Jointer，这个Jointer的连接关系：

逻辑引擎在解析编排图元件时，会给每一个元件端口创建一个Jointer实例：

1. //构造Jointers
2. for (const out of activityMeta.outPorts || []) {
3.    //出端口对应的Jointer
4.    activity.jointers.outputs.push(new Jointer(out.id, out.name))
5. }
6. for (const input of activityMeta.inPorts || []) {
7.    //入端口对应的Jointer
8.    activity.jointers.inputs.push(new Jointer(input.id, input.name))
9. }

复制代码

新创建的Jointer，它的下游是空的，就是说成员变量的outlets数组是空的，并没有挂接到真实的业务处理。要调用Jointer的connect方法，把Activity的处理函数作为下游连接过去。
最先想到的实现方式是Acitvity有一个inputHandler方法，根据端口名字分发数据到相应处理函数：

1. export interface IActivity<ConfigMeta = unknown> {
2.   id: string;
3.   //连接器，跟元件节点的端口异议对应
4.   jointers: IActivityJointers,
5.   //元件节点配置，每个Activity的配置都不一样，故而用泛型
6.   config?: ConfigMeta;
7.   //入口处理函数
8.   inputHandler(portName:string, inputValue: unknown, context?:unknown):void
9.   //销毁
10.   destory(): void;
11. }
13. export abstract class SomeActivity implements IActivity<SomeConfigMeta> {
14.   id: string;
15.   jointers: IActivityJointers;
16.   config?: SomeConfigMeta;
17.   constructor(public meta: INodeDefine<ConfigMeta>) {
18.     this.id = meta.id
19.     this.jointers = new ActivityJointers()
20.     this.config = meta.config;
21.   }
23.   //入口处理函数
24.   inputHandler(portName:string, inputValue: unknown, context?:unknown){
25.     switch(portName){
26.       case PORTNAME1:
27.         port1Handler(inputValue, context)
28.         break
29.       case PORTNAME2:
30.         ...
31.         break
32.       ...
33.     }
34.   }
36.   //端口1处理函数
37.   port1Handler = (inputValue: unknown, context?:unknown)=>{
38.     ...
39.   }
40.   
41.   destory = () => {
42.     //销毁处理
43.     ...
44.   }
45. }

复制代码

LogicFlow解析编排JSON，碰到SomeActivity对应的元件时，如下处理：

1. //创建SomeActivity实例
2. const someNode = new SomeActivity(meta)
4. //构造Jointers
5. for (const out of activityMeta.outPorts || []) {
6.    //出端口对应的Jointer
7.    activity.jointers.outputs.push(new Jointer(out.id, out.name))
8. }
9. for (const input of activityMeta.inPorts || []) {
10.    //入端口对应的Jointer
11.    const jointer = new Jointer(input.id, input.name)
12.    activity.jointers.inputs.push(jointer)
13.    //给入口对应的连接器，挂接输入处理函数
14.    jointer.connect(someNode.inputHandler)
15. }

复制代码

业务逻辑挂接到出端口

入口处理函数，处理完数据以后，需要调用出端口连接器的push方法，把数据分发出去：

具体实现代码：

1. export abstract class SomeActivity implements IActivity<SomeConfigMeta> {
2.   jointers: IActivityJointers;
3.   ...
4.   //入口处理函数
5.   inputHandler(portName:string, inputValue: unknown, context?:unknown){
6.     switch(portName){
7.       case PORTNAME1:
8.         port1Handler(inputValue, context)
9.         break
10.       case PORTNAME2:
11.         ...
12.         break
13.       ...
14.     }
15.   }
17.   //端口1处理函数
18.   port1Handler = (inputValue: unknown, context?:unknown)=>{
19.     ...
20.     //处理后得到新的值：newInputValue 和新的context：newContext
21.     //把数据分发到相应出口
22.     this.jointers.getOutput(somePortName).push(newInputValue, newContext)
23.   }
24.   ...
25. }

复制代码

入端口跟出端口，连贯起来，一个Activtity内部的流程就跑通了：

出端口挂接其它元件节点

入端口关联的是Activity的自身处理函数，出端口关联的是外部处理函数，这些外部处理函数有能是其它连接器（Jointer）的push方法，也可能来源于其它跟应用对接的部分。
如果是关联的是其他节点的Jointer，关联关系是通过逻辑编排图中的连线定义的。

解析器先构造完所有的节点，然后遍历一遍连线，调用连线源Jointer的conect方法，参数是目标Jointer的push，就把关联关系构建起来了：

1.     for (const lineMeta of this.flowMeta.lines) {
2.       //先找起始节点，这个后面会详细介绍，现在可以先忽略
3.       let sourceJointer = this.jointers.inputs.find(jointer => jointer.id === lineMeta.source.nodeId)
4.       if (!sourceJointer && lineMeta.source.portId) {
5.         sourceJointer = this.activities.find(reaction => reaction.id === lineMeta.source.nodeId)?.jointers?.outputs.find(output => output.id === lineMeta.source.portId)
6.       }
7.       if (!sourceJointer) {
8.         throw new Error("Can find source jointer")
9.       }
11.       //先找起终止点，这个后面会详细介绍，现在可以先忽略
12.       let targetJointer = this.jointers.outputs.find(jointer => jointer.id === lineMeta.target.nodeId)
13.       if (!targetJointer && lineMeta.target.portId) {
14.         targetJointer = this.activities.find(reaction => reaction.id === lineMeta.target.nodeId)?.jointers?.inputs.find(input => input.id === lineMeta.target.portId)
15.       }
17.       if (!targetJointer) {
18.         throw new Error("Can find target jointer")
19.       }
21.       //重点关注这里，把一条连线的首尾相连，构造起连接关系
22.       sourceJointer.connect(targetJointer.push)
23.     }

复制代码

特殊的元件节点：开始节点、结束节点

到目前为止，解析引擎部分，已经能够成功解析普通的元件并成功连线，但是一个编排的入口跟出口尚未处理，对应的是编排图的输入节点（开始节点）跟输出节点（结束节点）

这两个节点，没有任何业务逻辑，只是辅助把外部输入，连接到内部的元件；或者把内部的输出，发送给外部。所以，这两个节点，只是简单的Jointer就够了。
如果把一个逻辑编排看作一个元件节点：

输入元件节点对应的是输入端口，输出元件节点对应的是输出端口。既然逻辑编排也有自己端口，那么LogicFlow也要聚合ActivityJointers：

引擎解析的时候，要根据开始元件节点跟结束元件节点，构建LogicFlow的Jointer：

1. export class LogicFlow {
2.   id: string;
3.   jointers: IActivityJointers = new ActivityJointers();
4.   activities: IActivity[] = [];
6.   constructor(private flowMeta: ILogicFlowDefine) {
7.           ...
8.     //第一步，解析节点
9.     this.constructActivities()
10.     ...
11.   }
13.   //构建一个图的所有节点
14.   private constructActivities() {
15.     for (const activityMeta of this.flowMeta.nodes) {
16.       switch (activityMeta.type) {
17.         case NodeType.Start:
18.           //start只有一个端口，可能会变成其它流程的端口，所以name谨慎处理
19.           this.jointers.inputs.push(new Jointer(activityMeta.id, activityMeta.name || "input"));
20.           break;
21.         case NodeType.End:
22.           //end 只有一个端口，可能会变成其它流程的端口，所以name谨慎处理
23.           this.jointers.outputs.push(new Jointer(activityMeta.id, activityMeta.name || "output"));
24.           break;
25.       }
26.       ...
27.     }
28.   }
29. }

复制代码

经过这样的处理，一个逻辑编排就可以变成一个元件节点，被其他逻辑编排所引用，具体实现细节，本文后面再展开叙述。
根据元件节点创建Activity实例

在逻辑编排图中，一种类型的元件节点，在解析引擎中会对应一个实现了IActivity接口的类。比如，循环节点，对应LoopActivity；条件节点，对应ConditionActivity；调试节点，对应DebugActivity；拆分对象节点，对应SplitObjectActivity。
这些Activity要跟具体的元件节点建立一一对应关系，在DSL中以activityName作为关联枢纽。这样解析引擎根据activityName查找相应的Activity类，并创建实例。
工厂方法

如何找到并创建节点单元对应的Activity实例呢？最简单的实现方法，是给每个Activity类实现一个工厂方法，建立一个activityName跟工厂方法的映射map，解析引擎根据这个map实例化相应的Activity。简易代码：

1. //工厂方法的类型定义
2. export type ActivityFactory = (meta:ILogiFlowDefine)=>IActivity
4. //activityName跟工厂方法的映射map
5. export const activitiesMap:{[activityName:string]:ActivityFactory} = {}
7. export class LoopActivity implements IActivity{
8.   ...
9.   constructor(protected meta:ILogiFlowDefine){}
10.   inputHandler=(portName:string, inputValue:unknown, context:unknown)=>{
11.     if(portName === "input"){
12.       //逻辑处理
13.       ...
14.     }
15.   }
16.   ...
17. }
19. //LoopActivity的工厂方法
20. export const LoopActivityFactory:ActivityFactory = (meta:ILogiFlowDefine)=>{
21.   return new LoopActivity(meta)
22. }
24. //把工厂方法注册进map，跟循环节点的activityName对应好
25. activitiesMap["loop"] = LoopActivityFactory
27. //LogicFlow的解析代码
28. export class LogicFlow {
29.   id: string;
30.   jointers: IActivityJointers = new ActivityJointers();
31.   activities: IActivity[] = [];
33.   constructor(private flowMeta: ILogicFlowDefine) {
34.           ...
35.     //第一步，解析节点
36.     this.constructActivities()
37.     ...
38.   }
40.   //构建一个图的所有节点
41.   private constructActivities() {
42.     for (const activityMeta of this.flowMeta.nodes) {
43.       switch (activityMeta.type) {
44.               ...
45.         case NodeType.Activity:
46.           //查找元件节点对应的ActivityFactory
47.           const activityFactory = activitiesMap[activityMeta.activityName]
48.           if(activityFactory){
49.             //创建Activity实例
50.             this.activities.push(activityFactory(activityMeta))
51.           }else{
52.             //提示错误
53.           }
54.           break;
55.       }
56.       ...
57.     }
58.   }
59. }

复制代码

引入反射

正常情况下，上面的实现方法，已经够用了。但是，作为一款开放软件，会有大量的自定义Activity的需求。上面的实现方式，会让Activity的实现代码略显繁琐，并且所有的输入端口都要通过switch判断转发到相应处理函数。
我们希望把这部分工作推到框架层做，让具体Activity的实现更简单。所以，引入了Typescipt的反射机制：注解。通过注解自动注册Activity类，通过注解直接关联端口与相应的处理函数，省去switch代码。
代码经过改造以后，就变成这样：

1. //通过注解注册LoopActivity类
2. @Activity("loop")
3. export class LoopActivity implements IActivity{
4.   ...
5.   constructor(protected meta:ILogiFlowDefine){}
6.   //通过注解把input端口跟该处理函数关联
7.   @Input("input")
8.   inputHandler=(inputValue:unknown, context:unknown)=>{
9.     //逻辑处理
10.     ...
11.   }
12.   ...
13. }
16. //LogicFlow的解析代码
17. export class LogicFlow {
18.   id: string;
19.   jointers: IActivityJointers = new ActivityJointers();
20.   activities: IActivity[] = [];
22.   constructor(private flowMeta: ILogicFlowDefine) {
23.           ...
24.     //第一步，解析节点
25.     this.constructActivities()
26.     ...
27.   }
29.   //构建一个图的所有节点
30.   private constructActivities() {
31.     for (const activityMeta of this.flowMeta.nodes) {
32.       switch (activityMeta.type) {
33.               ...
34.         case NodeType.Activity:
35.           //根据反射拿到Activity的构造函数
36.           const activityContructor = ...//此处是反射代码
37.           if(activityContructor){
38.             //创建Activity实例
39.             this.activities.push(activityContructor(activityMeta))
40.           }else{
41.             //提示错误
42.           }
43.           break;
44.       }
45.       ...
46.     }
47.   }
48. }

复制代码

LogicFlow是框架层代码，用户不需要关心具体的实现细节。LoopActivity的代码实现，明显简洁了不少。
Input注解接受一个参数作为端口名称，参数默认值是input。
还有一种节点，它的输入端口是不固定的，可以动态增加或者删除。比如：

合并节点就是动态入口的节点，它的功能是接收入口传来的数据，等所有数据到齐以后，合并成一个对象转发到输出端口。这个节点，有异步等待的功能。
为了处理这种节点，我们引入新的注解DynamicInput。实际项目中合并节点Activity的完整实现：

1. import {
2.   AbstractActivity,
3.   Activity,
4.   DynamicInput
5. } from '@rxdrag/minions-runtime';
6. import { INodeDefine } from '@rxdrag/minions-schema';
8. @Activity(MergeActivity.NAME)
9. export class MergeActivity extends AbstractActivity<unknown> {
10.   public static NAME = 'system.merge';
11.   private noPassInputs: string[] = [];
12.   private values: { [key: string]: unknown } = {};
14.   constructor(meta: INodeDefine<unknown>) {
15.     super(meta);
16.     this.resetNoPassInputs();
17.   }
19.   @DynamicInput
20.   inputHandler = (inputName: string, inputValue: unknown) => {
21.     this.values[inputName] = inputValue;、
22.     //删掉已经收到数据的端口名
23.     this.noPassInputs = this.noPassInputs.filter(name=>name !== inputName)
24.     if (this.noPassInputs.length === 0) {
25.       //next方法，把数据转发到指定出口，第二个参数是端口名，默认值input
26.       this.next(this.values);
27.       this.resetNoPassInputs();
28.     }
29.   };
31.   resetNoPassInputs(){
32.     for (const input of this.meta.inPorts || []) {
33.       this.noPassInputs.push(input.name);
34.     }
35.   }
36. }

复制代码

注解DynamicInput不需要绑定固定的端口，所以就不需要输入端口的名称。
子编排的解析

子编排就是一段完整的逻辑编排，跟普通的逻辑编排没有任何区别。只是它需要被其它编排引入，这个引入是通过附加一个Activity实现的。

1. export interface ISubLogicFLowConfig {
2.   logicFlowId?: string
3. }
5. export interface ISubMetasContext{
6.   subMetas:ILogicFlowDefine[]
7. }
9. @Activity(SubLogicFlowActivity.NAME)
10. export class SubLogicFlowActivity implements IActivity {
11.   public static NAME = "system-react.subLogicFlow"
12.   id: string;
13.   jointers: IActivityJointers;
14.   config?: ISubLogicFLowConfig;
15.   logicFlow?: LogicFlow;
17.   //context可以从引擎外部注入的，此处不必纠结它是怎么来的这个细节
18.   constructor(meta: INodeDefine<ISubLogicFLowConfig>, context: ISubMetasContext) {
19.     this.id = meta.id
20.     //通过配置中的LogicFlowId，查找子编排对应的JSON数据
21.     const defineMeta = context?.subMetas?.find(subMeta => subMeta.id === meta.config?.logicFlowId)
22.     if (defineMeta) {
23.       //解析逻辑编排，new LogicFlow 就是解析一段逻辑编排，也可以在别处被调用
24.       this.logicFlow = new LogicFlow(defineMeta, context)
25.       //把解析后的连接器对应到本Activity
26.       this.jointers = this.logicFlow.jointers
27.     } else {
28.       throw new Error("No meta on sub logicflow")
29.     }
30.   }
31.   
32.   destory(): void {
33.     this.logicFlow?.destory();
34.     this.logicFlow = undefined;
35.   }
36. }

复制代码

因为不需要把端口绑定到相应的处理函数，故该Activity并没有使用Input相关注解。
嵌入式编排的解析

逻辑编排中，最复杂的部分，就是嵌入式编排的解析，希望小编能解释清楚。
再看一遍嵌入式编排的表现形式：

这是自定义循环节点。虽然它端口直接跟内部的编排节点相连，但是实际上这种情况是无法直接调用new LogicFlow 来解析内部逻辑编排的，需要进行转换。引擎解析的时候，把会把上面的子编排重组成如下形式：

首先，给子编排添加输入节点，名称跟ID分别对应自定义循环的入端口名称跟ID；添加输出节点，名称跟ID分别对应自定义循环的出端口名称跟ID。
然后，把一个图中的红色数字标注的连线，替换成第二个图中蓝色数字标注的连线。
容器节点的端口，并不会跟转换后的输入节点或者输出节点直接连接，而是在实现中根据业务逻辑适时调用，故用粗虚线表示。
自定义循环具体实现代码：

1. import { AbstractActivity, Activity, Input, LogicFlow } from "@rxdrag/minions-runtime";
2. import { INodeDefine } from "@rxdrag/minions-schema";
3. import _ from "lodash"
5. export interface IcustomizedLoopConifg {
6.   fromInput?: boolean,
7.   times?: number
8. }
10. @Activity(CustomizedLoop.NAME)
11. export class CustomizedLoop extends AbstractActivity<IcustomizedLoopConifg> {
12.   public static NAME = "system.customizedLoop"
13.   public static PORT_INPUT = "input"
14.   public static PORT_OUTPUT = "output"
15.   public static PORT_FINISHED = "finished"
16.   
17.   finished = false
19.   logicFlow?: LogicFlow;
21.   constructor(meta: INodeDefine<IcustomizedLoopConifg>) {
22.     super(meta)
23.     if (meta.children) {
24.       //通过portId关联子流程的开始跟结束节点，端口号对应节点号
25.       //此处的children是被引擎转换过处理的
26.       this.logicFlow = new LogicFlow({ ...meta.children, id: meta.id }, undefined)
28.       //把子编排的出口，挂接到本地处理函数
29.       const outputPortMeta = this.meta.outPorts?.find(
30.         port=>port.name === CustomizedLoop.PORT_OUTPUT
31.       )
32.       if(outputPortMeta?.id){
33.         this.logicFlow?.jointers?.getOutput(outputPortMeta?.name)?.connect(
34.           this.oneOutputHandler
35.         )
36.       }else{
37.         console.error("No output port in CustomizedLoop")
38.       }
40.       const finishedPortMeta = this.meta.outPorts?.find(
41.         port=>port.name === CustomizedLoop.PORT_FINISHED
42.       )
43.       if(finishedPortMeta?.id){
44.         this.logicFlow?.jointers?.getOutput(finishedPortMeta?.id)?.connect(
45.           this.finisedHandler
46.         )
47.       }else{
48.         console.error("No finished port in CustomizedLoop")
49.       }
50.       
51.     } else {
52.       throw new Error("No implement on CustomizedLoop meta")
53.     }
54.   }
56.   @Input()
57.   inputHandler = (inputValue?: unknown, context?:unknown) => {
58.     let count = 0
59.     if (this.meta.config?.fromInput) {
60.       if (!_.isArray(inputValue)) {
61.         console.error("Loop input is not array")
62.       } else {
63.         for (const one of inputValue) {
64.           //转发输入到子编排
65.           this.getInput()?.push(one, context)
66.           count++
67.           //如果子编排调用了结束
68.           if(this.finished){
69.             break
70.           }
71.         }
72.       }
73.     } else if (_.isNumber(this.meta.config?.times)) {
74.       for (let i = 0; i < (this.meta.config?.times || 0); i++) {
75.         //转发输入到子编排
76.         this.getInput()?.push(, context)
77.         count++
78.         //如果子编排调用了结束
79.         if(this.finished){
80.           break
81.         }
82.       }
83.     }
84.     //如果子编排中还没有被调用过finished
85.     if(!this.finished){
86.       this.next(count, CustomizedLoop.PORT_FINISHED, context)
87.     }
88.   }
90.   getInput(){
91.     return this.logicFlow?.jointers?.getInput(CustomizedLoop.PORT_INPUT)
92.   }
94.   oneOutputHandler = (value: unknown, context?:unknown)=>{
95.     //输出到响应端口
96.     this.output(value, context)
97.   }
99.   finisedHandler = (value: unknown, context?:unknown)=>{
100.     //标识已调用过finished
101.     this.finished = true
102.     //输出到响应端口
103.     this.next(value, CustomizedLoop.PORT_FINISHED, context)
104.   }
106.   output = (value: unknown, context?:unknown) => {
107.     this.next(value, CustomizedLoop.PORT_OUTPUT, context)
108.   }
109. }

复制代码

基础的逻辑编排引擎，基本全部介绍完了，清楚了节点之间的编排机制，是时候定义节点的连线规则了。
节点的连线规则

一个节点，是一个对象。有状态，有副作用。有状态的对象没有约束的互连，是非常危险的行为。
这种情况会面临一个诱惑，或者说用户自己也分不清楚。就是把节点当成无状态对象使用，或者直接认为节点就是无状态的，不加限制的把连线连到某个节点的入口上。
比如上面计算学生总分例子，可能会被糊涂的用户连成这样：

这种连接方式，直接造成收集数组节点无法正常工作。
逻辑编排之所以直观，在于它把每一个个数据流通的路径都展示出来了。在一个通路上的一个节点，最好只完成一个该通路的功能。另一个通路如果想完成同样的功能，最好再新建一个对象：

这样两个收集数组节点，就互不干扰了。
要实现这样的约束，只需要加一个连线规则：同一个入端口，只能连一条线。
有了这条规则，节点对象状态带来的不利影响，基本消除了。
在这样的规则下，收集数组节点的入口不能连接多条连线，只需要把它重新设计成如下形式：

一个出端口，可以往外连接多条连线，用于表示并行执行。另一条规则就是：同一个出端口，可以有多条连线。
数据是从左往右流动，所以再加上最后一条规则：入端口在节点左侧，出端口在节点右侧。
所有的连线规则完成了，蛮简单的，编辑器层面可以直接做约束，防止用户输错。
编辑器的实现

编辑器布局

整个编辑器分为图中标注的四个区域。

• ① 工具栏，编辑器常规操作，比如撤销、重做、删除等。
  
• ② 工具箱（物料箱），存放可以被拖放的元件物料，这些物料是可以从外部注入到编辑器的。
  
• ③ 画布区，绘制逻辑编排图的画布。每个节点都有自己的坐标，要基于这个对DSL进行扩展，给节点附加坐标信息。画布基于阿里antv X6实现。
  
• ④ 属性面板，编辑元件节点的配置信息。物料是从编辑器外部注入的，物料对应节点的配置是变化的，所以属性面板内的组件也是变化的，使用RxDrag的低代码渲染引擎来实现，外部注入的物料要写到相应的Schema信息。低代码Schema相关内容，请参考另一篇文章《实战，一个高扩展、可视化低代码前端，详实、完整》
  

扩展DSL

前面定义的DSL用在逻辑编排解析引擎里，足够了。但是，在画布上展示，还缺少节点位置跟尺寸信息。设计器画布是基于X6实现的，要添加X6需要的信息，来扩展DSL：

1. export interface IX6NodeDefine {
2.   /** 节点x坐标 */
3.   x: number;
4.   /** 节点y坐标  */
5.   y: number;
6.   /** 节点宽度 */
7.   width: number;
8.   /** 节点高度 */
9.   height: number;
10. }
11. // 扩展后节点
12. export interface IActivityNode extends INodeDefine {
13.   x6Node?: IX6NodeDefine
14. }

复制代码

这些信息，足以在画布上展示一个完整的逻辑编排图了。
元件物料定义

工具箱区域②跟画布区域③显示节点时，使用了共同的元素：元件图标，元件标题，图标颜色，这些可以放在物料的定义里。
物料还需要：元件对应的Acitvity名字，属性面板④ 的配置Schema。具体定义：

1. import { NodeType, IPortDefine } from "./dsl";
3. //端口定义
4. export interface IPorts {
5.   //入端口
6.   inPorts?: IPortDefine[];
7.   //出端口
8.   outPorts?: IPortDefine[];
9. }
11. //元件节点的物料定义
12. export interface IActivityMaterial<ComponentNode = unknown, NodeSchema = unknown, Config = unknown, MaterialContext = unknown> {
13.   //标题
14.   label: string;
15.   //节点类型，NodeType在DLS中定义，这里根据activityType决定画上的图形样式
16.   activityType: NodeType;
17.   //图标代码，react的话，相当于React.ReactNode
18.   icon?: ComponentNode;
19.   //图标颜色
20.   color?: string;
21.   //属性面板配置，可以适配不同的低代码Schema，使用RxDrag的话，这可以是INodeSchema类型
22.   schema?: NodeSchema;
23.   //默认端口，元件节点的端口设置的默认值，大部分节点端口跟默认值是一样的，
24.   //部分动态配置端口，会根据配置有所变化
25.   defaultPorts?: IPorts;
26.   //画布中元件节点显示的子标题
27.   subTitle?: (config?: Config, context?: MaterialContext) => string | undefined;
28.   //对应解析引擎里的Activity名称，根据这个名字实例化相应的节点业务逻辑对象
29.   activityName: string;
30. }
32. //物料分类，用于在工具栏上，以手风琴风格分组物料
33. export interface ActivityMaterialCategory<ComponentNode = unknown, NodeSchema = unknown, Config = unknown, MaterialContext = unknown> {
34.   //分类名
35.   name: string;
36.   //分类包含的物料
37.   materials: IActivityMaterial<ComponentNode, NodeSchema, Config, MaterialContext>[];
38. }

复制代码

只要符合这个定义的物料，都是可以被注入设计器的。
在前面定义DSL的时候， INodeDefine 也有一个一样的属性是 activityName。没错，这两个activityName指代的对象是一样的。画布渲染dsl的时候，会根据activityName查找相应的物料，根据物料携带的信息展示，入图标、颜色、属性配置组件等。
在做前端物料跟元件的时候，为了重构方便，会把activityName以存在Activity的static变量里，物料定义直接引用，端口名称也是类似的处理。看一个最简单的节点，Debug节点的代码。
Activity代码：
[code]import { Activity, Input, AbstractActivity } from "@rxdrag/minions-runtime"import { INodeDefine } from "@rxdrag/minions-schema"//调试节点配置export interface IDebugConfig {  //提示信息  tip?: string,  //是否已关闭  closed?: boolean}@Activity(DebugActivity.NAME)export class DebugActivity extends AbstractActivity {  //对应INodeDeifne 跟IActivityMaterial的 activityName  public static NAME = "system.debug"  constructor(meta: INodeDefine) {    super(meta)  }  //入口处理函数  @Input()  inputHandler(inputValue: unknown): void {    if (!this.config?.closed) {      console.log(`
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！