feat(bare-flow): allow to dynamically link dataflow

src/app/@dataflow/core/bare-flow.spec.ts

@@ -49,6 +49,50 @@ describe('BareFlow', () => {
 			})();
 			rst.deploy();
 
+			expectObservable(rst.getOutput()).toBe(expected, values);
+		});
+	});
+	it('prerequest twice(same value), but got once only', () => {
+		scheduler.run((helpers) => {
+			const { cold, hot, expectObservable, expectSubscriptions, flush } = helpers;
+			const values: { [id: string]: DataFlowNode } = {
+				a: [{ a: 555 }, []],
+				b: [{ b: 123 }, []],
+			};
+			const pre = cold('a--a-', values);
+			const expected = 'b----';
+
+			const rst = new (class extends BareFlow {
+				public prerequest$ = pre;
+				protected request(pre: DataFlowNode): Observable<DataFlowNode> {
+					return of(values.b);
+				}
+			})();
+			rst.deploy();
+
+			expectObservable(rst.getOutput()).toBe(expected, values);
+		});
+	});
+	it('prerequest twice(different value), got twice', () => {
+		scheduler.run((helpers) => {
+			const { cold, hot, expectObservable, expectSubscriptions, flush } = helpers;
+			const values: { [id: string]: DataFlowNode } = {
+				a: [{ ab: 555 }, []],
+				b: [{ ab: 123 }, []],
+				c: [{ ab: 556 }, []],
+				d: [{ ab: 124 }, []],
+			};
+			const pre = cold('a--b-', values);
+			const expected = 'c--d-';
+
+			const rst = new (class extends BareFlow {
+				public prerequest$ = pre;
+				protected request(pre: DataFlowNode): Observable<DataFlowNode> {
+					return of([{ ab: pre[0]['ab'] + 1 }, []]);
+				}
+			})();
+			rst.deploy();
+
 			expectObservable(rst.getOutput()).toBe(expected, values);
 		});
 	});

src/app/@dataflow/core/bare-flow.ts

@@ -1,5 +1,5 @@
 import { Observable, of } from 'rxjs';
-import { switchMap, take } from 'rxjs/operators';
+import { switchMap, take, tap, startWith, distinctUntilChanged, skip } from 'rxjs/operators';
 
 export type DataFlowNode = [object, Error[]];
 
@@ -8,19 +8,26 @@ export abstract class BareFlow {
 	protected abstract request(pre: DataFlowNode): Observable<DataFlowNode>;
 	private bareData$: Observable<DataFlowNode>;
 	private deployed = false;
+	private boostrapData: DataFlowNode;
 	public deploy() {
 		this.bareData$ = this.prerequest$.pipe(
 			switchMap(
 				(pre): Observable<DataFlowNode> => {
 					if (pre[1].length === 0) return this.request(pre).pipe(take(1));
 					return of(pre);
 				}
-			)
+			),
+			tap((x) => (this.boostrapData = x))
 		);
+		this.bareData$.pipe(take(1)).subscribe();
 		this.deployed = true;
 	}
 	public getOutput(): Observable<DataFlowNode> {
 		if (!this.deployed) throw new Error('run deploy before getOutput');
-		return this.bareData$;
+		return this.bareData$.pipe(
+			startWith(this.boostrapData),
+      distinctUntilChanged(),
+      skip(1) // don't why need it , otherwise, test failure. refs: https://stackoverflow.com/a/52157317
+    );
 	}
 }