[refactor] replace Call Bus with Stream Request for Async Execution

package.json

@@ -1,6 +1,6 @@
 {
     "name": "echarts-jsx",
-    "version": "1.1.2-rc.4",
+    "version": "1.2.0-rc.0",
     "license": "LGPL-3.0",
     "author": "[email protected]",
     "description": "A real JSX wrapper for ECharts based on TypeScript & Web components",
@@ -48,7 +48,7 @@
         "prettier": "^3.2.5",
         "rimraf": "^5.0.5",
         "typedoc": "^0.25.13",
-        "typedoc-plugin-mdn-links": "^3.1.23",
+        "typedoc-plugin-mdn-links": "^3.1.24",
         "typescript": "~5.4.5"
     },
     "prettier": {

source/Option.ts

@@ -4,8 +4,8 @@ import { CustomElement, toCamelCase } from 'web-utility';
 import { EChartsElement } from './renderers/core';
 import { ProxyElement } from './Proxy';
 import {
-    ZRElementEventName,
-    callBus,
+    ZRElementEventHandler,
+    streamRequest,
     unwrapEventHandler,
     wrapEventHandler
 } from './utility';
@@ -38,15 +38,7 @@ export abstract class ECOptionElement
 
     renderer?: EChartsElement;
 
-    constructor() {
-        super();
-
-        this.updateOption.start();
-        this.removeEventListener.start();
-        this.addEventListener.start();
-    }
-
-    async connectedCallback() {
+    connectedCallback() {
         super.connectedCallback();
 
         for (
@@ -60,11 +52,9 @@ export abstract class ECOptionElement
             throw new ReferenceError(
                 `<${this.tagName.toLowerCase()} /> should be append to a DOM tree within <ec-svg-renderer /> or <ec-canvas-renderer />`
             );
-        await this.renderer.ready;
-
-        this.updateOption.run();
-        this.removeEventListener.run();
-        this.addEventListener.run();
+        this.renderer.connectOption(this.emitOption.stream);
+        this.renderer.connectAddListener(this.emitAddListener.stream);
+        this.renderer.connectRemoveListener(this.emitRemoveListener.stream);
     }
 
     #nextTick?: Promise<void>;
@@ -78,28 +68,33 @@ export abstract class ECOptionElement
         });
     }
 
-    updateOption = callBus(() => {
+    emitOption = streamRequest<[EChartsOption]>();
+
+    updateOption() {
         const data = this.toJSON();
 
-        const option = this.isSeries
-            ? { series: [{ ...data, type: this.chartName }] }
-            : { [this.chartTagName]: data };
+        const option = (
+            this.isSeries
+                ? { series: [{ ...data, type: this.chartName }] }
+                : { [this.chartTagName]: data }
+        ) as EChartsOption;
 
-        this.renderer.setOption(option);
-    });
+        return this.emitOption(option);
+    }
 
-    addEventListener = callBus((name: string, handler: EventListener) => {
-        this.renderer.core.on(
-            name as ZRElementEventName,
+    emitAddListener = streamRequest<[string, string, ZRElementEventHandler]>();
+
+    addEventListener(name: string, handler: EventListener) {
+        return this.emitAddListener(
+            name,
             this.eventSelector,
             wrapEventHandler.call(this, name, handler)
         );
-    });
+    }
 
-    removeEventListener = callBus((event: string, handler: EventListener) => {
-        this.renderer.core.off(
-            event as ZRElementEventName,
-            unwrapEventHandler(handler)
-        );
-    });
+    emitRemoveListener = streamRequest<[string, ZRElementEventHandler]>();
+
+    removeEventListener(event: string, handler: EventListener) {
+        return this.emitRemoveListener(event, unwrapEventHandler(handler));
+    }
 }

source/renderers/core.ts

@@ -2,13 +2,15 @@ import { EChartsOption, ResizeOpts } from 'echarts';
 import { ECharts, init } from 'echarts/core';
 import { ECBasicOption } from 'echarts/types/dist/shared';
 import { debounce } from 'lodash';
+import { ReadableStream } from 'web-streams-polyfill';
 import { CustomElement, parseDOM } from 'web-utility';
 
 import { ProxyElement } from '../Proxy';
 import {
+    StreamRequest,
     ZRElementEventHandler,
     ZRElementEventName,
-    callBus,
+    streamRequest,
     unwrapEventHandler,
     wrapEventHandler
 } from '../utility';
@@ -30,21 +32,17 @@ export abstract class EChartsElement
     extends ProxyElement<EChartsElementState>
     implements CustomElement
 {
-    core?: ECharts;
+    #core?: ECharts;
     #coreDefer = Promise.withResolvers<void>();
 
-    get ready() {
-        return this.#coreDefer.promise;
-    }
-
     get renderer() {
         const [_, type] = this.tagName.toLowerCase().split('-');
 
         return type;
     }
 
     get options() {
-        return this.core.getOption();
+        return this.#core.getOption();
     }
 
     constructor() {
@@ -53,9 +51,6 @@ export abstract class EChartsElement
         this.attachShadow({ mode: 'open' }).append(
             parseDOM('<div style="height: 100%" />')[0]
         );
-        this.setOption.start();
-        this.removeEventListener.start();
-        this.addEventListener.start();
     }
 
     connectedCallback() {
@@ -71,49 +66,98 @@ export abstract class EChartsElement
     disconnectedCallback() {
         globalThis.removeEventListener?.('resize', this.handleResize);
 
-        this.core?.dispose();
+        this.#core?.dispose();
     }
 
     #init() {
         var { theme, initOptions, ...props } = this.toJSON();
 
-        this.core = init(
+        this.#core = init(
             this.shadowRoot.firstElementChild as HTMLDivElement,
             theme,
             { ...initOptions, renderer: this.renderer }
         );
         this.#coreDefer.resolve();
 
         this.setOption({ grid: {}, ...props });
-        this.setOption.run();
-        this.removeEventListener.run();
-        this.addEventListener.run();
-    }
 
-    setOption = callBus((data: EChartsOption) =>
-        this.core.setOption(data, false, true)
-    );
+        this.processStream(this.setOption.stream, (data: EChartsOption) => {
+            this.#core.setOption(data, false, true);
+        });
+        this.processStream(
+            this.removeEventListener.stream,
+            (event: string, handler: EventListener) => {
+                this.#core.getZr().off(event, unwrapEventHandler(handler));
+            }
+        );
+        this.processStream(
+            this.addEventListener.stream,
+            (name: string, handler: EventListener) => {
+                this.#core
+                    .getZr()
+                    .on(
+                        name as ZRElementEventName,
+                        wrapEventHandler.call(this, name, handler)
+                    );
+            }
+        );
+    }
+    setOption = streamRequest<[EChartsOption]>();
 
     setProperty(key: string, value: any) {
         super.setProperty(key, value);
 
         this.setOption(this.toJSON());
     }
 
-    addEventListener = callBus((name: string, handler: EventListener) => {
-        this.core
-            .getZr()
-            .on(
-                name as ZRElementEventName,
-                wrapEventHandler.call(this, name, handler)
-            );
-    });
+    protected async processStream<I extends any[], O = void>(
+        stream: ReadableStream<StreamRequest<I>>,
+        executor: (...input: I) => O,
+        context?: object
+    ) {
+        await this.#coreDefer.promise;
+
+        for await (const request of stream) {
+            const { input, output } = request as StreamRequest<I>;
+
+            try {
+                const data = executor.apply(context, input);
+
+                output.resolve(data);
+            } catch (error) {
+                output.reject(error);
+            }
+        }
+    }
+    protected connectChildStream<I extends any[], O = void>(
+        executor: (...input: I) => O
+    ) {
+        const that = this;
+
+        return function (stream: ReadableStream<StreamRequest<I>>) {
+            return that.processStream(stream, executor, this);
+        };
+    }
+
+    connectOption = this.connectChildStream((option: EChartsOption) =>
+        this.setOption(option)
+    );
+
+    connectAddListener = this.connectChildStream(
+        (event: string, selector: string, handler: ZRElementEventHandler) => {
+            this.#core.on(event as ZRElementEventName, selector, handler);
+        }
+    );
+    connectRemoveListener = this.connectChildStream(
+        (event: string, handler: ZRElementEventHandler) => {
+            this.#core.off(event as ZRElementEventName, handler);
+        }
+    );
+    addEventListener = streamRequest<[string, EventListener]>();
 
-    removeEventListener = callBus((event: string, handler: EventListener) => {
-        this.core.getZr().off(event, unwrapEventHandler(handler));
-    });
+    removeEventListener = streamRequest<[string, EventListener]>();
 
     handleResize = debounce(() =>
-        this.core.resize(this.toJSON().resizeOptions)
+        this.#core.resize(this.toJSON().resizeOptions)
     );
 }

source/utility.ts

@@ -4,63 +4,33 @@ import {
     ReadableStreamDefaultController
 } from 'web-streams-polyfill';
 
-export interface QueueTask {
-    context: object;
-    input: any[];
-    output: PromiseWithResolvers<any>;
+export interface StreamRequest<I extends any[], O = void> {
+    context?: object;
+    input: I;
+    output: PromiseWithResolvers<O>;
 }
 
-export interface CallBusWrapper<T> {
-    (...data: any[]): Promise<T>;
-}
-
-export interface CallBusHandlers {
-    start: () => Promise<void>;
-    run: () => void;
-}
-
-export function callBus<T>(worker: (...data: any[]) => T) {
-    const clutch = Promise.withResolvers<void>();
-
-    var handler: ReadableStreamDefaultController<QueueTask>;
+export function streamRequest<I extends any[], O = void>() {
+    var handler: ReadableStreamDefaultController<StreamRequest<I, O>>;
 
-    const stream = new ReadableStream<QueueTask>({
+    const stream = new ReadableStream<StreamRequest<I, O>>({
         start: controller => {
             handler = controller;
         }
     });
 
-    function addTask(context: object, input: any[]) {
-        const task = {
-            context,
+    function call(...input: I) {
+        const request = {
+            context: this,
             input,
-            output: Promise.withResolvers<T>()
+            output: Promise.withResolvers<O>()
         };
-        handler.enqueue(task);
+        handler.enqueue(request);
 
-        return task;
+        return request.output.promise;
     }
 
-    const run = () => clutch.resolve();
-
-    async function start() {
-        await clutch.promise;
-
-        for await (const { context, input, output } of stream)
-            try {
-                const data = await worker.apply(context, input);
-
-                output.resolve(data);
-            } catch (error) {
-                output.reject(error);
-            }
-    }
-    return Object.assign<CallBusWrapper<T>, CallBusHandlers>(
-        function (...input) {
-            return addTask(this, input).output.promise;
-        },
-        { start, run }
-    );
+    return Object.assign(call, { stream });
 }
 
 export const EventKeyPattern = /^on(\w+)/;