feat: add demo

MoeOtaLab · Jan 14, 2024 · bc1b375 · bc1b375
1 parent cd6f49a
commit bc1b375
Show file tree

Hide file tree

Showing 13 changed files with 1,003 additions and 22 deletions.
diff --git a/packages/core/src/operators/index.ts b/packages/core/src/operators/index.ts
@@ -3,7 +3,14 @@
 // 之后可以区分为操作流和数据流
 
 // ================ streams operators ================= //
-export { constValue, sum, proxyData, combine, transform } from './streams';
+export {
+  constValue,
+  sum,
+  proxyData,
+  combine,
+  transform,
+  merge,
+} from './streams';
 
 // ================ values operators ================= //
 export { sumValue } from './values';

diff --git a/packages/core/src/operators/streams/combine.ts b/packages/core/src/operators/streams/combine.ts
@@ -10,7 +10,7 @@ export function combine<
 ) {
   const result = new ModelState<any[]>([]);
 
-  streamSource.forEach((input, index) => {
+  streamSource.forEach((input) => {
     input.subscribe((_value, extraInfo) => {
       const allValue = [...streamSource, ...appendSource].map(
         (item) => item.current,

diff --git a/packages/core/src/operators/streams/index.ts b/packages/core/src/operators/streams/index.ts
@@ -3,3 +3,4 @@ export { proxyData } from './proxyData';
 export { sum } from './sum';
 export { combine } from './combine';
 export { transform } from './transform';
+export { merge } from './merge';
diff --git a/packages/core/src/operators/streams/merge.ts b/packages/core/src/operators/streams/merge.ts
@@ -0,0 +1,16 @@
+import { type Atom } from '../../ModelState/State';
+import { ModelState } from '../..';
+
+export function merge<ObservableSubjectList extends Array<Atom<any>>>(
+  streamSource: ObservableSubjectList,
+) {
+  const result = new ModelState<any[]>([]);
+
+  streamSource.forEach((input) => {
+    input.subscribe((value, extraInfo) => {
+      result.update(value, extraInfo.concat('combine'));
+    });
+  });
+
+  return result;
+}
diff --git a/packages/core/src/operators/streams/transform.ts b/packages/core/src/operators/streams/transform.ts
@@ -8,8 +8,13 @@ export function transform<StreamValue, ResultValue>(
   return steam.pipe((observer) => {
     const result = new ModelState<ResultValue | undefined>(undefined);
 
-    observer.subscribe((value, extra) => {
-      result.update(action(value), extra.concat('transform'));
+    observer.subscribe(async (value, extra) => {
+      const actionResult = action(value);
+      let resultValue = actionResult;
+      if (actionResult instanceof Promise) {
+        resultValue = await actionResult;
+      }
+      result.update(resultValue, extra.concat('transform'));
     });
 
     return result;

diff --git a/packages/flow/src/Diagrams/Compiler/graph.ts b/packages/flow/src/Diagrams/Compiler/graph.ts
@@ -106,7 +106,14 @@ export class NodeGraph {
     for (const edge of this.edges) {
       if (edge.source && edge.target) {
         const currentDegree = degreeMap.get(edge.target) || 0;
-        degreeMap.set(edge.target, currentDegree + 1);
+        const targetNode = this.nodeMap.get(edge.target);
+        const ignoreDegreeIds = targetNode
+          ? getOperatorFromNode(targetNode)?.getIgnoreDegreeIds?.(targetNode) ||
+            []
+          : [];
+        if (!ignoreDegreeIds.includes(edge.targetHandle || '')) {
+          degreeMap.set(edge.target, currentDegree + 1);
+        }
       }
     }
 

diff --git a/packages/flow/src/Diagrams/Operators/CombineOperator/index.tsx b/packages/flow/src/Diagrams/Operators/CombineOperator/index.tsx
@@ -94,26 +94,33 @@ export class CombineOperator
       this.getHintPorts(node, 'output')?.find((item) => item.hint === 'output')
         ?.id || '';
 
-    const mainInputSet = new Set(
-      this.getMainInputPorts(node)?.map((item) => item.id),
-    );
-    const appendInputSet = new Set(
-      this.getAppendInputPorts(node)?.map((item) => item.id),
-    );
+    const mainInputList =
+      this.getMainInputPorts(node)?.map((item) => item.id) || [];
+
+    const appendInputList =
+      this.getAppendInputPorts(node)?.map((item) => item.id) || [];
 
     const sourceItems = nodeGraph.findSourceNodes(node.id) || [];
-    const mainInputSourceIds = sourceItems
-      .filter((item) => mainInputSet.has(item.handleId))
-      .map((item) => item.relatedHandleId);
+    const sourceItemMap = new Map(
+      sourceItems.map((item) => [item.handleId, item]),
+    );
+
+    const mainInputSourceIds = mainInputList
+      .map((id) => sourceItemMap.get(id))
+      .map((item) => item?.relatedHandleId);
 
-    const appendInputSourceIds = sourceItems
-      .filter((item) => appendInputSet.has(item.handleId))
-      .map((item) => item.relatedHandleId);
+    const appendInputSourceIds = appendInputList
+      .map((id) => sourceItemMap.get(id))
+      .map((item) => item?.relatedHandleId);
 
     return [
       `const ${formatVariableName(handleId)} = ${EosOperatorsSymbol}.combine(
-        [${mainInputSourceIds.map((id) => formatVariableName(id)).join(',')}],
-        [${appendInputSourceIds.map((id) => formatVariableName(id)).join(',')}]
+        [${mainInputSourceIds
+          .map((id) => (id ? formatVariableName(id) : 'undefined'))
+          .join(',')}],
+        [${appendInputSourceIds
+          .map((id) => (id ? formatVariableName(id) : 'undefined'))
+          .join(',')}]
       )`,
     ];
   }

diff --git a/packages/flow/src/Diagrams/Operators/MergeOperator/index.tsx b/packages/flow/src/Diagrams/Operators/MergeOperator/index.tsx
@@ -0,0 +1,141 @@
+import { type Node } from 'reactflow';
+import { NodeTypeEnum } from '../../Nodes/NodeTypeEnum';
+import { MetaOperator } from '../Operator';
+import { EndPoint, type IMergeOperatorData } from '../types';
+import { type IGenerationOption } from '../../Compiler/graph';
+import { EosCoreSymbol, EosOperatorsSymbol } from '../../Compiler/runtime';
+
+export class MergeOperator
+  extends MetaOperator<IMergeOperatorData>
+  implements MetaOperator<IMergeOperatorData>
+{
+  nodeColor?: string | undefined = '#C21292';
+
+  constructor() {
+    super({
+      operatorName: 'Merge',
+      operatorType: 'MergeOperator',
+      nodeType: NodeTypeEnum.Node,
+    });
+  }
+
+  create(): Node<IMergeOperatorData<Record<string, any>>> {
+    return super.create({
+      endPointOptions: {
+        endPointList: [
+          new EndPoint({
+            type: 'group',
+            hint: 'output',
+            children: [
+              new EndPoint({
+                label: 'output',
+                hint: 'output',
+                variableType: 'array',
+                type: 'source',
+              }),
+            ],
+          }),
+          new EndPoint({
+            type: 'group',
+            hint: 'input',
+            label: 'Stream',
+            defaultChildData: {
+              hint: 'input',
+              type: 'target',
+              ignoreDegree: true,
+            },
+            allowAddAndRemoveChildren: true,
+            children: [
+              new EndPoint({
+                hint: 'input',
+                type: 'target',
+                ignoreDegree: true,
+              }),
+            ],
+          }),
+        ],
+      },
+    });
+  }
+
+  getHintPorts(node: Node<IMergeOperatorData>, hint: string) {
+    return node.data.endPointOptions?.endPointList?.find(
+      (item) => item.hint === hint,
+    )?.children;
+  }
+
+  getIgnoreDegreeIds(
+    node: Node<IMergeOperatorData<Record<string, any>>>,
+  ): string[] {
+    const mainInputList =
+      this.getHintPorts(node, 'input')?.map((item) => item.id) || [];
+    return mainInputList;
+  }
+
+  generateBlockDeclarations(
+    options: IGenerationOption<IMergeOperatorData<Record<string, any>>>,
+  ): string[] {
+    const { node, formatVariableName } = options;
+
+    const handleId =
+      this.getHintPorts(node, 'output')?.find((item) => item.hint === 'output')
+        ?.id || '';
+
+    const mainInputList =
+      this.getHintPorts(node, 'input')?.map((item) => item.id) || [];
+    return [
+      ...mainInputList.map(
+        (inputId) =>
+          `const ${formatVariableName(
+            inputId,
+          )} = new ${EosCoreSymbol}.ModelState(undefined)`,
+      ),
+      `const ${formatVariableName(handleId)} = ${EosOperatorsSymbol}.merge(
+        [${mainInputList.map((id) => formatVariableName(id)).join(',')}],
+      )`,
+    ];
+  }
+
+  generateBlockOutput(
+    options: IGenerationOption<IMergeOperatorData<Record<string, any>>>,
+  ): string[] {
+    return [];
+  }
+
+  generateBlockRelation(
+    options: IGenerationOption<IMergeOperatorData<Record<string, any>>>,
+  ): string[] {
+    const { node, formatVariableName, nodeGraph } = options;
+
+    const mainInputList =
+      this.getHintPorts(node, 'input')?.map((item) => item.id) || [];
+
+    const sourceItems = nodeGraph.findSourceNodes(node.id) || [];
+    const sourceItemMap = new Map(
+      sourceItems.map((item) => [item.handleId, item]),
+    );
+
+    const relationList = mainInputList
+      .map((inputId) => ({
+        inputId,
+        targetId: sourceItemMap.get(inputId)?.relatedHandleId || '',
+      }))
+      .filter((item) => item.inputId && item.targetId);
+
+    return [
+      ...relationList.map(
+        ({ inputId, targetId }) => `${formatVariableName(
+          targetId,
+        )}.subscribe((val, extraInfo) => {
+          ${formatVariableName(
+            inputId,
+          )}.update(val, extraInfo.concat('${JSON.stringify({
+            currentNodeId: node.id,
+            fromPortId: targetId,
+            toPortId: inputId,
+          })}'));
+        });`,
+      ),
+    ];
+  }
+}
diff --git a/packages/flow/src/Diagrams/Operators/Operator.tsx b/packages/flow/src/Diagrams/Operators/Operator.tsx
@@ -115,11 +115,15 @@ export abstract class MetaOperator<
   }
 
   // ============ START: Code Generation ============= //
-  abstract generateBlockDeclarations(options: IGenerationOption): string[];
+  abstract generateBlockDeclarations(options: IGenerationOption<T>): string[];
 
-  abstract generateBlockRelation(options: IGenerationOption): string[];
+  abstract generateBlockRelation(options: IGenerationOption<T>): string[];
 
-  abstract generateBlockOutput(options: IGenerationOption): string[];
+  abstract generateBlockOutput(options: IGenerationOption<T>): string[];
+
+  getIgnoreDegreeIds(node: Node<T>): string[] {
+    return [];
+  }
 
   // ============ START: Hooks handler ============= //
   onAfterCreate(options: IHookOption<Node<T>>): void {}

diff --git a/packages/flow/src/Diagrams/Operators/index.ts b/packages/flow/src/Diagrams/Operators/index.ts
@@ -5,6 +5,7 @@ import { SumOperator } from './SumOperator';
 import { CustomOperator } from './CustomOperator';
 import { CombineOperator } from './CombineOperator';
 import { TransformOperator, EffectOperator } from './TransformOperator';
+import { MergeOperator } from './MergeOperator';
 
 import { registerOperators } from './OperatorMap';
 
@@ -18,6 +19,7 @@ registerOperators([
   new CombineOperator(),
   new TransformOperator(),
   new EffectOperator(),
+  new MergeOperator(),
 ]);
 
 export {

diff --git a/packages/flow/src/Diagrams/Operators/types/index.ts b/packages/flow/src/Diagrams/Operators/types/index.ts
@@ -54,6 +54,7 @@ export class EndPoint {
   allowAddAndRemoveChildren?: boolean;
   variableType?: 'array' | 'object' | 'primary';
   defaultChildData?: Partial<EndPoint>;
+  ignoreDegree?: boolean;
 
   constructor(data: Partial<EndPoint>) {
     Object.assign(this, data);
@@ -143,3 +144,9 @@ export interface IEffectOperatorData<
 > extends ITranformOperatorData<NodeOptions> {
   // noop
 }
+
+export interface IMergeOperatorData<
+  NodeOptions extends Record<string, any> = Record<string, any>,
+> extends IMetaOperatorData<NodeOptions> {
+  // noop
+}