refactor: 客户端加载

packages/loader/src/index.ts

@@ -1 +1,4 @@
+export * from './progress';
 export * from './stream';
+export * from './task';
+export * from './types';

packages/loader/src/progress.ts

@@ -0,0 +1,91 @@
+import { clamp } from 'lodash-es';
+import { ILoadProgressTotal, ILoadTask, LoadDataType } from './types';
+import { logger } from '@motajs/common';
+
+export class LoadProgressTotal<
+    T extends LoadDataType = LoadDataType,
+    R = any
+> implements ILoadProgressTotal<T, R> {
+    /** 当前已经附着的加载任务 */
+    private readonly attached: Map<ILoadTask<T, R>, number> = new Map();
+    /** 当前已经加载完毕的任务 */
+    readonly loadedTasks: Set<ILoadTask<T, R>> = new Set();
+    /** 当前已经添加的任务 */
+    readonly addedTasks: Set<ILoadTask<T, R>> = new Set();
+
+    /** 总加载量 */
+    private total: number = 0;
+    /** 当前已经加载的字节数 */
+    private loaded: number = 0;
+
+    /** 下一次触发 `onProgress` 时兑现 */
+    private nextPromise: Promise<void>;
+    /** 兑现当前的 `nextPromise` */
+    private nextResolve: () => void;
+
+    async *[Symbol.asyncIterator]() {
+        while (true) {
+            if (this.loadedTasks.size === this.addedTasks.size) {
+                return;
+            }
+            yield this.nextPromise;
+        }
+    }
+
+    constructor() {
+        const { promise, resolve } = Promise.withResolvers<void>();
+        this.nextPromise = promise;
+        this.nextResolve = resolve;
+    }
+
+    addTask(task: ILoadTask<T, R>) {
+        this.addedTasks.add(task);
+    }
+
+    onProgress(task: ILoadTask<T, R>, loaded: number, total: number): void {
+        if (!this.addedTasks.has(task)) {
+            logger.warn(95);
+            return;
+        }
+        if (!this.attached.has(task)) {
+            this.total += total;
+        }
+        if (task.contentLoaded) {
+            this.loadedTasks.add(task);
+        }
+        const before = this.attached.getOrInsert(task, 0);
+        if (total !== 0) {
+            this.loaded += loaded - before;
+        }
+        this.attached.set(task, loaded);
+        this.nextResolve();
+        const { promise, resolve } = Promise.withResolvers<void>();
+        this.nextPromise = promise;
+        this.nextResolve = resolve;
+    }
+
+    getLoadedByte(): number {
+        return this.loaded;
+    }
+
+    getTotalByte(): number {
+        return this.loaded;
+    }
+
+    getLoadedTasks(): number {
+        return this.loadedTasks.size;
+    }
+
+    getAddedTasks(): number {
+        return this.addedTasks.size;
+    }
+
+    getTaskRatio(): number {
+        return this.loadedTasks.size / this.addedTasks.size;
+    }
+
+    getByteRatio(): number {
+        if (this.total === 0) return 0;
+        return clamp(this.loaded / this.total, 0, 1);
+    }
+}

packages/loader/src/stream.ts

@@ -1,76 +1,15 @@
 import { logger } from '@motajs/common';
-import EventEmitter from 'eventemitter3';
+import { IStreamLoader, IStreamReader } from './types';
 
-export interface IStreamController<T = void> {
-    readonly loading: boolean;
-
-    /**
-     * 开始流传输
-     */
-    start(): Promise<T>;
-
-    /**
-     * 主动终止流传输
-     * @param reason 终止原因
-     */
-    cancel(reason?: string): void;
-}
-
-export interface IStreamReader<T = any> {
-    /**
-     * 接受字节流流传输的数据
-     * @param data 传入的字节流数据，只包含本分块的内容
-     * @param done 是否传输完成
-     */
-    pump(
-        data: Uint8Array | undefined,
-        done: boolean,
-        response: Response
-    ): Promise<void>;
-
-    /**
-     * 当前对象被传递给加载流时执行的函数
-     * @param controller 传输流控制对象
-     */
-    piped(controller: IStreamController<T>): void;
-
-    /**
-     * 开始流传输
-     * @param stream 传输流对象
-     * @param controller 传输流控制对象
-     */
-    start(
-        stream: ReadableStream,
-        controller: IStreamController<T>,
-        response: Response
-    ): Promise<void>;
-
-    /**
-     * 结束流传输
-     * @param done 是否传输完成，如果为 false 的话，说明可能是由于出现错误导致的终止
-     * @param reason 如果没有传输完成，那么表示失败的原因
-     */
-    end(done: boolean, reason?: string): void;
-}
-
-interface StreamLoaderEvent {
-    data: [data: Uint8Array | undefined, done: boolean];
-}
-
-export class StreamLoader
-    extends EventEmitter<StreamLoaderEvent>
-    implements IStreamController<void>
-{
+export class StreamLoader implements IStreamLoader {
     /** 传输目标 */
     private target: Set<IStreamReader> = new Set();
     /** 读取流对象 */
-    private stream?: ReadableStream;
+    private stream: ReadableStream | null = null;
 
     loading: boolean = false;
 
-    constructor(public readonly url: string) {
-        super();
-    }
+    constructor(public readonly url: string) {}
 
     /**
      * 将加载流传递给字节流读取对象
@@ -83,7 +22,14 @@ export class StreamLoader
         }
         this.target.add(reader);
         reader.piped(this);
-        return this;
+    }
+
+    unpipe(reader: IStreamReader): void {
+        if (this.loading) {
+            logger.warn(46);
+            return;
+        }
+        this.target.delete(reader);
     }
 
     async start() {

packages/loader/src/task.ts

@@ -0,0 +1,158 @@
+import { sumBy } from 'lodash-es';
+import {
+    ILoadDataTypeMap,
+    ILoadTask,
+    ILoadTaskInit,
+    ILoadTaskProcessor,
+    ILoadTaskProgress,
+    LoadDataType,
+    RequestMethod
+} from './types';
+
+/** 文字解码 */
+const loadTextDecoder = new TextDecoder();
+
+export class LoadTask<T extends LoadDataType, R> implements ILoadTask<T, R> {
+    readonly dataType: T;
+    readonly identifier: string;
+    readonly url: string | URL;
+    readonly processor: ILoadTaskProcessor<T, R>;
+    readonly progress: ILoadTaskProgress<T, R>;
+    readonly method?: RequestMethod;
+    readonly body?: BodyInit;
+    readonly headers?: HeadersInit;
+
+    contentLoaded: boolean = false;
+    loadedByte: number = 0;
+    totalByte: number = 0;
+
+    /** 加载的 `Promise` */
+    private readonly loadPromise: Promise<R>;
+    /** 兑现加载对象 */
+    private readonly loadResolve: (data: R) => void;
+
+    /** 加载结果 */
+    private loadedData: R | null = null;
+
+    constructor(init: ILoadTaskInit<T, R>) {
+        this.dataType = init.dataType;
+        this.identifier = init.identifier;
+        this.url = this.resolveURL(init.url);
+        this.processor = init.processor;
+        this.progress = init.progress;
+        this.method = init.method;
+        this.body = init.body;
+        this.headers = init.headers;
+
+        const { promise, resolve } = Promise.withResolvers<R>();
+        this.loadPromise = promise;
+        this.loadResolve = resolve;
+    }
+
+    private resolveURL(url: string | URL) {
+        if (typeof url === 'string') {
+            return `${import.meta.env.BASE_URL}${url}`;
+        } else {
+            return url;
+        }
+    }
+
+    private processUnstreamableResponse(
+        response: Response
+    ): Promise<ILoadDataTypeMap[T]> {
+        switch (this.dataType) {
+            case LoadDataType.ArrayBuffer:
+                return response.arrayBuffer();
+            case LoadDataType.Blob:
+                return response.blob();
+            case LoadDataType.JSON:
+                return response.json();
+            case LoadDataType.Text:
+                return response.text();
+            case LoadDataType.Uint8Array:
+                return response.bytes();
+        }
+    }
+
+    private processStreamChunkResponse(
+        chunks: Uint8Array<ArrayBuffer>[]
+    ): ILoadDataTypeMap[T] {
+        if (this.dataType === LoadDataType.Blob) {
+            return new Blob(chunks);
+        }
+        const totalLength = sumBy(chunks, value => value.length);
+        const stacked: Uint8Array<ArrayBuffer> = new Uint8Array(totalLength);
+        let offset = 0;
+        for (let i = 0; i < chunks.length; i++) {
+            stacked.set(chunks[i], offset);
+            offset += chunks[i].length;
+        }
+        switch (this.dataType) {
+            case LoadDataType.ArrayBuffer:
+                return stacked.buffer;
+            case LoadDataType.Uint8Array:
+                return stacked;
+        }
+        const text = loadTextDecoder.decode(stacked);
+        switch (this.dataType) {
+            case LoadDataType.Text:
+                return text;
+            case LoadDataType.JSON:
+                return JSON.parse(text);
+        }
+    }
+
+    private async processResponse(response: Response) {
+        const reader = response.body?.getReader();
+        const contentLength = response.headers.get('Content-Length') ?? '0';
+        const total = parseInt(contentLength, 10);
+        this.loadedByte = 0;
+        this.totalByte = total;
+        this.progress.onProgress(this, 0, total);
+        if (!reader) {
+            const data = await this.processUnstreamableResponse(response);
+            this.loadedByte = this.totalByte;
+            this.contentLoaded = true;
+            this.progress.onProgress(this, this.loadedByte, this.totalByte);
+            const processed = await this.processor.process(data, this);
+            this.loadedData = processed;
+            this.loadResolve(processed);
+            return;
+        }
+        let received = 0;
+        const chunks: Uint8Array<ArrayBuffer>[] = [];
+        while (true) {
+            const { done, value } = await reader.read();
+            if (value) {
+                chunks.push(value);
+                received += value.byteLength;
+            }
+            if (done) this.contentLoaded = true;
+            this.loadedByte = received;
+            this.progress.onProgress(this, received, total);
+            if (done) break;
+        }
+        const data = this.processStreamChunkResponse(chunks);
+        const processed = await this.processor.process(data, this);
+        this.loadedData = processed;
+        this.loadResolve(processed);
+    }
+
+    async start(): Promise<void> {
+        const response = await fetch(this.url, {
+            method: this.method,
+            body: this.body,
+            headers: this.headers
+        });
+        this.processResponse(response);
+        return;
+    }
+
+    loaded(): Promise<R> {
+        return this.loadPromise;
+    }
+
+    getLoadedData(): R | null {
+        return this.loadedData;
+    }
+}

packages/loader/src/types.ts

@@ -0,0 +1,228 @@
+//#region 流传输
+
+export interface IStreamController {
+    /** 当前是否正在加载 */
+    readonly loading: boolean;
+
+    /**
+     * 开始流传输
+     */
+    start(): Promise<void>;
+
+    /**
+     * 主动终止流传输
+     * @param reason 终止原因
+     */
+    cancel(reason?: string): void;
+}
+
+export interface IStreamReader {
+    /**
+     * 接受字节流流传输的数据
+     * @param data 传入的字节流数据，只包含本分块的内容
+     * @param done 是否传输完成
+     */
+    pump(
+        data: Uint8Array | undefined,
+        done: boolean,
+        response: Response
+    ): Promise<void>;
+
+    /**
+     * 当前对象被传递给加载流时执行的函数
+     * @param controller 传输流控制对象
+     */
+    piped(controller: IStreamController): void;
+
+    /**
+     * 当前对象取消指定加载流传输时执行的函数
+     * @param controller 传输流控制对象
+     */
+    unpiped(controller: IStreamController): void;
+
+    /**
+     * 开始流传输
+     * @param stream 传输流对象
+     * @param controller 传输流控制对象
+     */
+    start(
+        stream: ReadableStream,
+        controller: IStreamController,
+        response: Response
+    ): Promise<void>;
+
+    /**
+     * 结束流传输
+     * @param done 是否传输完成，如果为 false 的话，说明可能是由于出现错误导致的终止
+     * @param reason 如果没有传输完成，那么表示失败的原因
+     */
+    end(done: boolean, reason?: string): void;
+}
+
+export interface IStreamLoader extends IStreamController {
+    /**
+     * 将加载流传递给字节流读取对象
+     * @param reader 字节流读取对象
+     */
+    pipe(reader: IStreamReader): void;
+
+    /**
+     * 取消一个字节流读取对象的绑定
+     * @param reader 字节流读取对象
+     */
+    unpipe(reader: IStreamReader): void;
+}
+
+//#endregion
+
+//#region 加载任务
+
+export const enum LoadDataType {
+    ArrayBuffer,
+    Uint8Array,
+    Blob,
+    Text,
+    JSON
+}
+
+export interface ILoadDataTypeMap {
+    [LoadDataType.ArrayBuffer]: ArrayBuffer;
+    [LoadDataType.Uint8Array]: Uint8Array<ArrayBuffer>;
+    [LoadDataType.Blob]: Blob;
+    [LoadDataType.Text]: string;
+    [LoadDataType.JSON]: any;
+}
+
+export interface ILoadTaskProcessor<T extends LoadDataType, R> {
+    /**
+     * 处理加载内容
+     * @param response 处理前加载结果
+     * @param task 加载任务对象
+     */
+    process(response: ILoadDataTypeMap[T], task: ILoadTask<T, R>): Promise<R>;
+}
+
+export interface ILoadTaskProgress<T extends LoadDataType, R> {
+    /**
+     * 更新加载进度
+     * @param task 加载任务对象
+     * @param loaded 已加载的字节数
+     * @param total 文件总计字节数，如果此值为零说明无法读取到 `Content-Length`
+     */
+    onProgress(task: ILoadTask<T, R>, loaded: number, total: number): void;
+}
+
+export const enum RequestMethod {
+    GET = 'GET',
+    POST = 'POST',
+    HEAD = 'HEAD',
+    PUT = 'PUT',
+    DELETE = 'DELETE',
+    CONNECT = 'CONNECT',
+    OPTIONS = 'OPTIONS',
+    TRACE = 'TRACE',
+    PATCH = 'PATCH'
+}
+
+export interface ILoadTaskInit<T extends LoadDataType, R> {
+    /** 请求响应格式 */
+    readonly dataType: T;
+    /** 加载任务标识符 */
+    readonly identifier: string;
+    /** 加载目标 URL */
+    readonly url: string | URL;
+    /** 加载的处理对象，用于处理加载结果等 */
+    readonly processor: ILoadTaskProcessor<T, R>;
+    /** 加载进度对象，用于监控加载进度 */
+    readonly progress: ILoadTaskProgress<T, R>;
+    /** 请求模式 */
+    readonly method?: RequestMethod;
+    /** 请求体 */
+    readonly body?: BodyInit;
+    /** 请求头 */
+    readonly headers?: HeadersInit;
+}
+
+export interface ILoadTask<T extends LoadDataType, R> extends ILoadTaskInit<
+    T,
+    R
+> {
+    /** 当前是否加载完毕 */
+    readonly contentLoaded: boolean;
+    /** 已经加载的字节数 */
+    readonly loadedByte: number;
+    /** 该加载任务的总体字节数 */
+    readonly totalByte: number;
+
+    /**
+     * 开始此加载计划，返回一个 `Promise`，当得到服务器的响应后兑现
+     */
+    start(): Promise<void>;
+
+    /**
+     * 返回一个 `Promise`，当本计划加载完毕后兑现，兑现结果是加载结果
+     */
+    loaded(): Promise<R>;
+
+    /**
+     * 获取加载完成后的加载结果
+     */
+    getLoadedData(): R | null;
+}
+
+//#endregion
+
+//#region 内置组件
+
+export interface ILoadProgressTotal<
+    T extends LoadDataType = LoadDataType,
+    R = any
+> extends ILoadTaskProgress<T, R> {
+    /** 已经添加的加载任务对象 */
+    readonly addedTasks: Set<ILoadTask<T, R>>;
+    /** 当前已经加载完毕的任务对象 */
+    readonly loadedTasks: Set<ILoadTask<T, R>>;
+
+    /**
+     * 迭代加载进度，当 `yield` 的值被兑现时，说明加载进度更新
+     */
+    [Symbol.asyncIterator](): AsyncGenerator<void, void, void>;
+
+    /**
+     * 向该进度监听器添加加载任务对象
+     * @param task 加载任务对象
+     */
+    addTask(task: ILoadTask<T, R>): void;
+
+    /**
+     * 获取总体已加载的字节数
+     */
+    getLoadedByte(): number;
+
+    /**
+     * 获取总体需要加载的字节数
+     */
+    getTotalByte(): number;
+
+    /**
+     * 获取已经加载的字节数与总体需要加载的字节数之比
+     */
+    getByteRatio(): number;
+
+    /**
+     * 获取已经加载完毕的加载任务数量
+     */
+    getLoadedTasks(): number;
+
+    /**
+     * 获取此进度监听器已经添加的加载任务对象
+     */
+    getAddedTasks(): number;
+
+    /**
+     * 获取已经加载完毕的任务数量与已添加的加载任务数量之比
+     */
+    getTaskRatio(): number;
+}
+
+//#endregion