chore: 订阅数据

packages/li-editor/src/editor/editor.tsx

@@ -1,6 +1,7 @@
 import EventEmitter from '@antv/event-emitter';
 import type { AssetPackage } from '@antv/li-sdk';
 import { LIRuntimeApp } from '@antv/li-sdk';
+import { useUpdateEffect } from 'ahooks';
 import React, { useMemo } from 'react';
 import { Registry_Default_Editor_Widgets } from '../constants';
 import { LIEditorStateContext } from '../hooks';
@@ -133,6 +134,11 @@ class LIEditor extends EventEmitter {
         return { state: editorState, updateState: updateEditorState };
       }, [editorState, updateEditorState]);
 
+      // 同步 datasetSchemas 更新到 editorDatasetManager
+      useUpdateEffect(() => {
+        editorService.editorDatasetManager.update(editorState.datasets);
+      }, [editorState.datasets]);
+
       editorService.editorStateRef = editorState;
 
       return (

packages/li-editor/src/hooks/useEditorDatasets.ts

@@ -1,6 +1,5 @@
 import { isLocalDatasetSchema, isRemoteDatasetSchema } from '@antv/li-sdk';
-import { useUpdateEffect } from 'ahooks';
-import { useMemo, useState } from 'react';
+import { useMemo, useState, useSyncExternalStore } from 'react';
 import type { EditorDataset } from '../types';
 import { useEditorService } from './useEditor';
 import { useEditorState } from './useEditorState';
@@ -42,11 +41,6 @@ export const _useEditorDatasets_ = (datasetIds?: string[]) => {
   const { editorService } = useEditorService();
   const { editorDatasetManager } = editorService;
   const { state } = useEditorState();
-  const { serviceCache, datasets: datasetSchemas } = state;
-
-  useUpdateEffect(() => {
-    editorDatasetManager.update(datasetSchemas);
-  }, [datasetSchemas]);
 
   const [editorDatasets, setEditorDatasets] = useState(() => {});
 
@@ -56,15 +50,13 @@ export const _useEditorDatasets_ = (datasetIds?: string[]) => {
 export const _useEditorDatasets = () => {
   const { editorService } = useEditorService();
   const { editorDatasetManager } = editorService;
-  const { state } = useEditorState();
-  const { serviceCache, datasets: datasetSchemas } = state;
-
-  const [editorDatasets, setEditorDatasets] = useState(() => editorDatasetManager.getDatasetList());
 
-  useUpdateEffect(() => {
-    editorDatasetManager.update(datasetSchemas);
-  }, [datasetSchemas]);
+  useSyncExternalStore(
+    (onStoreChange) => editorDatasetManager.subscribe(onStoreChange),
+    () => editorDatasetManager.getSnapshot(),
+  );
 
+  const editorDatasets = editorDatasetManager.getSnapshot();
   const loading = editorDatasetManager.loading;
 
   return { editorDatasets, loading };

packages/li-editor/src/services/editor-dataset-manager.ts

@@ -1,7 +1,13 @@
 import type { DatasetField, DatasetSchema, DatasetServiceParams, RemoteDatasetSchema } from '@antv/li-sdk';
-import { getDatasetColumns, isLocalDatasetSchema, isLocalOrRemoteDatasetSchema } from '@antv/li-sdk';
-import { queryServiceClient } from '@antv/li-sdk/dist/esm/utils';
-import { QueryObserver } from '@tanstack/query-core';
+import {
+  getDatasetColumns,
+  isLocalDatasetSchema,
+  isLocalOrRemoteDatasetSchema,
+  isRemoteDatasetSchema,
+} from '@antv/li-sdk';
+import { queryServiceClient, Subscribable } from '@antv/li-sdk/dist/esm/utils';
+import type { QueryObserverOptions } from '@tanstack/query-core';
+import { QueriesObserver } from '@tanstack/query-core';
 import type { FieldPair, GeoField } from '../types';
 import { getGeoFields, getPointFieldPairs } from '../utils/dataset';
 import type AppService from './app-service';
@@ -77,27 +83,33 @@ export class EditorDataset {
   }
 }
 
+type EditorDatasetManagerListener = (data: EditorDataset[]) => void;
+
 /**
  * 编辑器的数据集状态管理器
  */
-class EditorDatasetManager {
+class EditorDatasetManager extends Subscribable<EditorDatasetManagerListener> {
   private datasets: Map<string, EditorDataset>;
   private appService: AppService;
+  private queriesObserver: QueriesObserver;
 
   constructor(appService: AppService, datasets: DatasetSchema[] = []) {
+    super();
     this.appService = appService;
     this.datasets = new Map(
       datasets.map((item) => {
         return [item.id, new EditorDataset(item)];
       }),
     );
+    this.queriesObserver = new QueriesObserver(queryServiceClient, this.getQueriesOptions(datasets));
+    this.queriesObserver.subscribe(this.onQueriesStoreChange);
   }
 
   /**
    * 数据集是否请求中，动态数据源类型情况
    */
   get loading() {
-    return this.getDatasetList().some((item) => item.loading);
+    return this.queriesObserver.getCurrentResult().some((item) => item.isLoading);
   }
 
   public getDatasetById(id: string) {
@@ -135,50 +147,95 @@ class EditorDatasetManager {
     for (let index = 0; index < datasets.length; index++) {
       const datasetSchema = datasets[index];
       const editorDataset = this.datasets.get(datasetSchema.id);
+
+      // 更新情况
       if (editorDataset && datasetSchema !== editorDataset.schema) {
-        // TODO: 什么情况下调 updateData
-        // 1. 基本元数据信息发生变更；2. 属性发生变更(columns\properties)；3. 数据发生变更；4. 类型发生变更
-        // 原则：只考虑会影响数据发生更新的情况；替换数据集的情况？
+        // 什么情况下 schema 会更新：
+        // 1. 基本元数据信息发生变更；
+        // 2. 属性发生变更(columns\properties)；
+        // 3. 数据发生变更；
+        // 4. 类型发生变更
+        // 原则：只考虑会影响数据发生更新的情况；也就是替换数据集的情况
         this.copyEditorDatasetAndUpdate(editorDataset, { schema: datasetSchema });
       } else {
-        // appService
+        // 新增情况
         this.datasets.set(datasetSchema.id, new EditorDataset(datasetSchema));
+        // TODO：新增数据集实现自动生成 layer
       }
     }
+
+    const datasetsSchema = this.getDatasetList().map((item) => item.schema);
+    this.queriesObserver.setQueries(this.getQueriesOptions(datasetsSchema));
+
+    this.notify();
   }
 
   /**
+   * 获取 QueriesObserver 请求参数
    * 动态数据源类型情况，异步请求数据
    */
-  public requestRemoteDataset(service: (...params: any) => Promise<any>) {
-    return service;
+  private getQueriesOptions(datasets: DatasetSchema[]) {
+    const queriesOptions: QueryObserverOptions[] = datasets
+      .filter((item): item is RemoteDatasetSchema => isRemoteDatasetSchema(item))
+      .map((dataset) => this.getQueryOptions(dataset));
+    return queriesOptions;
   }
 
-  private getRemoteDatasetService(serviceName: string, datasetSchema: RemoteDatasetSchema) {
+  /**
+   * 获取 QueryObserver 请求参数
+   * 动态数据源类型情况，异步请求数据
+   */
+  private getQueryOptions(datasetSchema: RemoteDatasetSchema) {
+    const { serviceType: serviceName, filter, properties } = datasetSchema;
     const datasetService = this.appService.getImplementDatasetService(serviceName);
     const service = datasetService.service;
-    const { filter, properties } = datasetSchema;
-    const queryObserver = new QueryObserver(queryServiceClient, {
+
+    const options: QueryObserverOptions = {
       queryKey: [serviceName, filter, properties],
       queryFn: (context) => {
         const serviceParams: DatasetServiceParams = { filter, properties, signal: context.signal };
         return service(serviceParams);
       },
+    };
+
+    return options;
+  }
+
+  protected onSubscribe(): void {
+    if (this.listeners.size === 1) {
+      // 订阅时，才发起数据请求
+    }
+  }
+
+  protected onUnsubscribe(): void {
+    if (!this.listeners.size) {
+      this.listeners = new Set();
+    }
+  }
+
+  private notify() {
+    this.listeners.forEach((listener) => {
+      listener(this.getDatasetList());
     });
+  }
 
-    const unsubscribe = queryObserver.subscribe((result) => {
-      console.log(result);
-      unsubscribe();
+  /**
+   * 数据请求结构变更
+   */
+  private onQueriesStoreChange() {
+    const queriesResult = this.queriesObserver.getCurrentResult();
+    queriesResult.forEach((queriesResult) => {
+      // queriesResult
     });
+  }
 
-    // if (service) {
-    //   service.service({
-    //     // properties: dataset.properties,
-    //     // TODO: filters 初始值问题
-    //   });
-    // }
+  public getSnapshot() {
+    return this.getDatasetList();
+  }
 
-    return service;
+  destroy() {
+    this.listeners = new Set();
+    this.queriesObserver.destroy();
   }
 }
 

packages/li-sdk/src/utils/service.ts

@@ -30,3 +30,37 @@ const queryClientConfig: QueryClientConfig = {
   },
 };
 export const queryServiceClient = new QueryClient(queryClientConfig);
+
+type Listener = () => void;
+
+export class Subscribable<TListener extends Function = Listener> {
+  protected listeners: Set<TListener>;
+
+  constructor() {
+    this.listeners = new Set();
+    this.subscribe = this.subscribe.bind(this);
+  }
+
+  subscribe(listener: TListener): () => void {
+    this.listeners.add(listener);
+
+    this.onSubscribe();
+
+    return () => {
+      this.listeners.delete(listener);
+      this.onUnsubscribe();
+    };
+  }
+
+  hasListeners(): boolean {
+    return this.listeners.size > 0;
+  }
+
+  protected onSubscribe(): void {
+    // Do nothing
+  }
+
+  protected onUnsubscribe(): void {
+    // Do nothing
+  }
+}