Revert "订阅调整"

This reverts commit 0bbe65b042.

graphic-pixi

@@ -1 +1 @@
-Subproject commit 91eb4a1d0835a8bc007febaf9054291c27d065ff
+Subproject commit 7fe73a8334f36cf917255a99d75c454abcb96d79

package.json

@@ -19,7 +19,6 @@
     "@pixi/graphics-extras": "^7.2.4",
     "@quasar/extras": "^1.0.0",
     "@stomp/stompjs": "^7.0.0",
-    "centrifuge": "^4.0.1",
     "axios": "^1.2.1",
     "google-protobuf": "^3.21.2",
     "js-base64": "^3.7.5",

src/drawApp/lineApp.ts

@@ -1,4 +1,11 @@
-import { GraphicApp, GraphicData, GraphicState } from 'src/jl-graphic';
+import {
+  GraphicApp,
+  GraphicData,
+  StompCli,
+  AppWsMsgBroker,
+  GraphicState,
+  GraphicIdGenerator,
+} from 'src/jl-graphic';
 import {
   TrainData,
   TrainOperateInteraction,
@@ -52,6 +59,7 @@ import { SeparatorTemplate } from 'src/graphics/separator/Separator';
 import { SeparatorData } from './graphics/SeparatorInteraction';
 
 let lineApp: GraphicApp | null = null;
+let msgBroker: AppWsMsgBroker | null = null;
 
 import { ContextMenu } from 'src/jl-graphic/ui/ContextMenu';
 import { MenuItemOptions } from 'src/jl-graphic/ui/Menu';
@@ -87,6 +95,9 @@ export function destroyLineApp(): void {
     lineApp.destroy();
     lineApp = null;
   }
+  if (msgBroker) {
+    msgBroker.close();
+  }
 }
 
 export function initLineApp(dom: HTMLElement): GraphicApp {
@@ -218,11 +229,12 @@ export async function loadLineDatas(app: GraphicApp) {
       QuickJumpMenu.open(e.global);
     });
 
-    app.enableWsMassaging({
+    StompCli.new({
       wsUrl: `${getWebsocketUrl()}`,
       token: getJwtToken() as string,
     });
-    app.subscribe({
+    msgBroker = new AppWsMsgBroker(app);
+    msgBroker.subscribe({
       destination: `/queue/line/${lineId}/device`,
       messageConverter: (message: Uint8Array) => {
         const states: GraphicState[] = [];
@@ -245,7 +257,7 @@ export async function loadLineDatas(app: GraphicApp) {
         return states;
       },
     });
-    app.subscribe({
+    msgBroker.subscribe({
       destination: `/queue/line/${lineId}/train`,
       messageConverter: (message: Uint8Array) => {
         const states: GraphicState[] = [];
@@ -272,7 +284,7 @@ export async function loadLineDatas(app: GraphicApp) {
       }
     });
     const lineNetStore = useLineNetStore();
-    app.subscribe({
+    msgBroker.subscribe({
       destination: '/queue/xian/ncc/alert',
       messageHandle: (message: Uint8Array) => {
         const storage = alert.NccAlertInfoMessage.deserialize(message);

src/drawApp/lineNetApp.ts

@@ -1,4 +1,10 @@
-import { GraphicApp, GraphicData, GraphicState } from 'src/jl-graphic';
+import {
+  GraphicApp,
+  GraphicData,
+  StompCli,
+  AppWsMsgBroker,
+  GraphicState,
+} from 'src/jl-graphic';
 import { getPublishLineNet } from 'src/api/PublishApi';
 import { graphicData } from 'src/protos/stationLayoutGraphics';
 import { state } from 'src/protos/ws_message';
@@ -31,6 +37,7 @@ import { alert } from 'src/protos/alertInfo';
 import { QNotifyUpdateOptions, Notify } from 'quasar';
 
 let lineNetApp: GraphicApp | null = null;
+let msgBroker: AppWsMsgBroker | null = null;
 
 export function getLineNetApp(): GraphicApp | null {
   return lineNetApp;
@@ -41,6 +48,9 @@ export function destroyLineNetApp(): void {
     lineNetApp.destroy();
     lineNetApp = null;
   }
+  if (msgBroker) {
+    msgBroker.close();
+  }
 }
 
 export function initLineNetApp(dom: HTMLElement): GraphicApp {
@@ -107,11 +117,12 @@ export async function loadLineNetDatas(app: GraphicApp) {
     pathLineList.forEach((pathLine) => {
       pathLine.visible = false;
     });
-    app.enableWsMassaging({
+    StompCli.new({
       wsUrl: `${getWebsocketUrl()}`,
       token: getJwtToken() as string,
     });
-    app.subscribe({
+    msgBroker = new AppWsMsgBroker(app);
+    msgBroker.subscribe({
       destination: '/queue/lineNet',
       messageConverter: (message: Uint8Array) => {
         const storage = state.WsLineNetMessage.deserialize(message);
@@ -122,7 +133,7 @@ export async function loadLineNetDatas(app: GraphicApp) {
         return states;
       },
     });
-    app.subscribe({
+    msgBroker.subscribe({
       destination: '/queue/xian/ncc/alert',
       messageHandle: (message: Uint8Array) => {
         const storage = alert.NccAlertInfoMessage.deserialize(message);

src/jl-graphic/app/JlGraphicApp.ts

@@ -21,10 +21,10 @@ import {
 import { AbsorbablePosition } from '../graphic';
 import {
   AppWsMsgBroker,
-  WsMsgCli,
+  StompCli,
   type AppStateSubscription,
-  type MessageCliOption,
+  type StompCliOption,
-} from '../message';
+} from '../message/WsMsgBroker';
 import { OperationRecord } from '../operation/JlOperation';
 import {
   AnimationManager,
@@ -498,10 +498,10 @@ export class GraphicApp extends EventEmitter<GraphicAppEvents> {
   }
 
   /**
-   * 启动websocket消息客户端
+   * 使能websocket Stomp通信
    */
-  enableWsMassaging(options: MessageCliOption) {
+  enableWsStomp(options: StompCliOption) {
-    WsMsgCli.new(options);
+    StompCli.new(options);
     this.wsMsgBroker = new AppWsMsgBroker(this);
   }
 
@@ -513,7 +513,7 @@ export class GraphicApp extends EventEmitter<GraphicAppEvents> {
       // console.log('APP订阅', sub)
       this.wsMsgBroker.subscribe(sub);
     } else {
-      throw new Error('订阅消息需先启动消息代理, 执行app.enableWebsocket()');
+      throw new Error('请先打开StompClient, 执行app.enableWebsocket()');
     }
   }
   /**

src/jl-graphic/message/BasicMessageClient.ts

@@ -1,40 +0,0 @@
-/* eslint-disable @typescript-eslint/no-explicit-any */
-import EventEmitter from 'eventemitter3';
-import { IMessageClient } from './MessageBroker';
-
-export type MessageHandler = (data: any) => void;
-
-export interface MessageClientEvents {
-  connecting: [ctx: any];
-  connected: [ctx: any];
-  disconnected: [ctx: any];
-}
-
-/**
- * 订阅接口
- */
-export interface ISubscription {
-  /**
-   * 取消订阅
-   */
-  unsubscribe(): void;
-}
-
-export abstract class MessageClient
-  extends EventEmitter<MessageClientEvents>
-  implements IMessageClient
-{
-  /**
-   * 订阅消息
-   * @param destination
-   * @param handle
-   */
-  abstract subscribe(
-    destination: string,
-    handle: MessageHandler
-  ): ISubscription;
-
-  abstract get connected(): boolean;
-
-  abstract close(): void;
-}

src/jl-graphic/message/CentrifugeBroker.ts

@@ -1,70 +0,0 @@
-import { Centrifuge, State } from 'centrifuge';
-import CentrifugeProtobuf from 'centrifuge/build/protobuf';
-import { MessageCliOption } from './MessageBroker';
-import {
-  ISubscription,
-  MessageClient,
-  MessageHandler,
-} from './BasicMessageClient';
-
-export class CentrifugeMessagingClient extends MessageClient {
-  options: MessageCliOption;
-  cli: Centrifuge;
-  constructor(options: MessageCliOption) {
-    super();
-    this.options = options;
-    if (this.options.protocol === 'json') {
-      this.cli = new Centrifuge(options.wsUrl, {
-        token: options.token,
-        protocol: options.protocol,
-      });
-    } else {
-      this.cli = new CentrifugeProtobuf(options.wsUrl, {
-        token: options.token,
-        protocol: options.protocol,
-      });
-    }
-
-    this.cli
-      .on('connecting', (ctx) => {
-        console.debug(`centrifuge连接中: ${ctx}, ${ctx.reason}`);
-        this.emit('connecting', ctx);
-      })
-      .on('connected', (ctx) => {
-        console.debug(`连接成功: ${ctx.transport}`);
-        this.emit('connected', ctx);
-      })
-      .on('disconnected', (ctx) => {
-        console.log(`断开连接: ${ctx.code}, ${ctx.reason}`);
-        this.emit('disconnected', ctx);
-      })
-      .on('error', (ctx) => {
-        console.error('centrifuge错误', ctx);
-      })
-      .connect();
-  }
-
-  get connected(): boolean {
-    return this.cli.state === State.Connected;
-  }
-
-  subscribe(destination: string, handle: MessageHandler): ISubscription {
-    const sub = this.cli.newSubscription(destination);
-    sub
-      .on('publication', (ctx) => {
-        if (this.options.protocol === 'json') {
-          console.log('收到centrifuge消息:', ctx.data);
-        }
-        handle(ctx.data);
-      })
-      .on('subscribed', (ctx) => {
-        console.log('订阅centrifuge服务消息成功', destination, ctx);
-      })
-      .subscribe();
-    return sub;
-  }
-
-  close(): void {
-    this.cli.disconnect();
-  }
-}

src/jl-graphic/message/MessageBroker.ts

@@ -1,263 +0,0 @@
-import EventEmitter from 'eventemitter3';
-import { CentrifugeMessagingClient } from './CentrifugeBroker';
-import { StompMessagingClient } from './WsMsgBroker';
-import { GraphicState } from '../core';
-import { GraphicApp } from '../app';
-import {
-  ISubscription,
-  MessageClientEvents,
-  MessageHandler,
-} from './BasicMessageClient';
-
-export enum ClientEngine {
-  Stomp,
-  Centrifugo,
-}
-
-export interface MessageCliOption {
-  /**
-   * 客户端引擎
-   */
-  engine?: ClientEngine;
-  /**
-   * 消息协议,默认protobuf
-   */
-  protocol?: 'json' | 'protobuf';
-  /**
-   * websocket url地址
-   */
-  wsUrl: string;
-  /**
-   * 认证token
-   */
-  token?: string;
-  // /**
-  //  * 认证失败处理
-  //  * @returns
-  //  */
-  // onAuthenticationFailed?: () => void;
-  // /**
-  //  * 连接成功处理
-  //  * @param ctx
-  //  * @returns
-  //  */
-  // onConnected?: (ctx: unknown) => void;
-  // /**
-  //  * 端口连接处理
-  //  */
-  // onDisconnected?: (ctx: unknown) => void;
-  // // 重连延时，默认3秒,设置为0不重连.
-  // reconnectDelay?: number;
-  // // 服务端过来的心跳间隔，默认30秒
-  // heartbeatIncoming?: number;
-  // // 到服务端的心跳间隔，默认30秒
-  // heartbeatOutgoing?: number;
-}
-
-const DefaultStompOption: MessageCliOption = {
-  engine: ClientEngine.Stomp,
-  protocol: 'protobuf',
-  wsUrl: '',
-  token: '',
-  // reconnectDelay: 3000,
-  // heartbeatIncoming: 30000,
-  // heartbeatOutgoing: 30000,
-};
-
-export interface IMessageClient extends EventEmitter<MessageClientEvents> {
-  /**
-   * 订阅消息
-   * @param destination
-   * @param handle
-   */
-  subscribe(destination: string, handle: MessageHandler): ISubscription;
-
-  /**
-   * 是否已经连接
-   */
-  get connected(): boolean;
-
-  /**
-   * 关闭连接
-   */
-  close(): void;
-}
-
-export class WsMsgCli {
-  private static client: IMessageClient;
-  private static options: MessageCliOption;
-  private static appMsgBroker: AppWsMsgBroker[] = [];
-  static new(options: MessageCliOption) {
-    if (WsMsgCli.client) {
-      // 已经创建
-      return;
-    }
-    WsMsgCli.options = Object.assign({}, DefaultStompOption, options);
-    if (WsMsgCli.options.engine == ClientEngine.Centrifugo) {
-      WsMsgCli.client = new CentrifugeMessagingClient(WsMsgCli.options);
-    } else {
-      WsMsgCli.client = new StompMessagingClient(WsMsgCli.options);
-      // WsMsgCli.client.onStompError = (frame: Frame) => {
-      // const errMsg = frame.headers['message'];
-      // if (errMsg === '401') {
-      //   console.warn('认证失败,断开WebSocket连接');
-      //   StompCli.close();
-      //   if (StompCli.options.onAuthenticationFailed) {
-      //     StompCli.options.onAuthenticationFailed();
-      //   }
-      // } else {
-      //   console.error('收到Stomp错误消息', frame);
-      // }
-      // };
-    }
-    const cli = WsMsgCli.client;
-    cli.on('connected', () => {
-      WsMsgCli.emitConnectStateChangeEvent(true);
-      WsMsgCli.appMsgBroker.forEach((broker) => {
-        broker.resubscribe();
-      });
-    });
-    cli.on('disconnected', () => {
-      WsMsgCli.emitConnectStateChangeEvent(false);
-    });
-  }
-
-  static emitConnectStateChangeEvent(connected: boolean) {
-    WsMsgCli.appMsgBroker.forEach((broker) => {
-      broker.app.emit('websocket-connect-state-change', connected);
-    });
-  }
-
-  static isConnected(): boolean {
-    return WsMsgCli.client && WsMsgCli.client.connected;
-  }
-
-  static trySubscribe(
-    destination: string,
-    handler: MessageHandler
-  ): ISubscription | undefined {
-    if (WsMsgCli.isConnected()) {
-      return WsMsgCli.client.subscribe(destination, handler);
-    }
-    return undefined;
-  }
-
-  static registerAppMsgBroker(broker: AppWsMsgBroker) {
-    WsMsgCli.appMsgBroker.push(broker);
-  }
-
-  static removeAppMsgBroker(broker: AppWsMsgBroker) {
-    const index = WsMsgCli.appMsgBroker.findIndex((mb) => mb == broker);
-    if (index >= 0) {
-      WsMsgCli.appMsgBroker.splice(index, 1);
-    }
-  }
-
-  static hasAppMsgBroker(): boolean {
-    return WsMsgCli.appMsgBroker.length > 0;
-  }
-
-  /**
-   * 关闭websocket连接
-   */
-  static close() {
-    if (WsMsgCli.client) {
-      WsMsgCli.client.close();
-    }
-  }
-}
-
-// 状态订阅消息转换器
-export type GraphicStateMessageConvert = (
-  message: Uint8Array
-) => GraphicState[];
-
-// 订阅消息处理器
-export type SubscriptionMessageHandle = (message: Uint8Array) => void;
-
-// 图形app状态订阅
-export interface AppStateSubscription {
-  /**
-   * 订阅路径
-   */
-  destination: string;
-  /**
-   * 图形状态消息转换
-   */
-  messageConverter?: GraphicStateMessageConvert;
-  /**
-   * 订阅消息处理
-   */
-  messageHandle?: SubscriptionMessageHandle;
-  /**
-   * 订阅成功对象，用于取消订阅
-   * 非客户端使用
-   */
-  subscription?: ISubscription;
-}
-
-/**
- * 图形APP的websocket消息代理
- */
-export class AppWsMsgBroker {
-  app: GraphicApp;
-  subscriptions: Map<string, AppStateSubscription> = new Map<
-    string,
-    AppStateSubscription
-  >();
-
-  constructor(app: GraphicApp) {
-    this.app = app;
-    WsMsgCli.registerAppMsgBroker(this);
-  }
-
-  subscribe(sub: AppStateSubscription) {
-    this.unsbuscribe(sub.destination); // 先尝试取消之前订阅
-    sub.subscription = WsMsgCli.trySubscribe(sub.destination, (data) => {
-      if (sub.messageConverter) {
-        const graphicStates = sub.messageConverter(data);
-        this.app.handleGraphicStates(graphicStates);
-      } else if (sub.messageHandle) {
-        sub.messageHandle(data);
-      } else {
-        console.error(
-          `订阅destination:${sub.destination}没有消息处理器或图形状态消息转换器`
-        );
-      }
-    });
-    this.subscriptions.set(sub.destination, sub);
-  }
-
-  /**
-   * 重连后重新订阅
-   */
-  resubscribe() {
-    this.subscriptions.forEach((record) => {
-      this.subscribe(record);
-    });
-  }
-
-  unsbuscribe(destination: string) {
-    const oldSub = this.subscriptions.get(destination);
-    if (oldSub) {
-      if (oldSub.subscription && WsMsgCli.isConnected()) {
-        oldSub.subscription.unsubscribe();
-      }
-      oldSub.subscription = undefined;
-    }
-  }
-
-  unsbuscribeAll() {
-    this.subscriptions.forEach((record) => {
-      this.unsbuscribe(record.destination);
-    });
-  }
-
-  /**
-   * 取消所有订阅，从通用Stomp客户端移除此消息代理
-   */
-  close() {
-    WsMsgCli.removeAppMsgBroker(this);
-    this.unsbuscribeAll();
-  }
-}

src/jl-graphic/message/WsMsgBroker.ts

@@ -1,79 +1,246 @@
-import { Client as StompClient, type Frame } from '@stomp/stompjs';
-import { MessageCliOption } from './MessageBroker';
 import {
-  ISubscription,
+  Client as StompClient,
-  MessageClient,
+  StompSubscription,
-  MessageHandler,
+  type Frame,
-} from './BasicMessageClient';
+  type Message,
+  type messageCallbackType,
+} from '@stomp/stompjs';
+import type { GraphicApp } from '../app/JlGraphicApp';
+import { GraphicState } from '../core/JlGraphic';
 
-const ReconnectDelay = 3000;
+export interface StompCliOption {
-const HeartbeatIncoming = 30000;
+  /**
-const HeartbeatOutgoing = 30000;
+   * websocket url地址
+   */
+  wsUrl: string;
+  /**
+   * 认证token
+   */
+  token?: string;
+  /**
+   * 认证失败处理
+   * @returns
+   */
+  onAuthenticationFailed?: () => void;
+  reconnectDelay?: number; // 重连延时，默认3秒,设置为0不重连.
+  heartbeatIncoming?: number; // 服务端过来的心跳间隔，默认30秒
+  heartbeatOutgoing?: number; // 到服务端的心跳间隔，默认30秒
+}
 
-export class StompMessagingClient extends MessageClient {
+const DefaultStompOption: StompCliOption = {
-  options: MessageCliOption;
+  wsUrl: '',
-  cli: StompClient;
+  token: '',
-  constructor(options: MessageCliOption) {
+  reconnectDelay: 3000,
-    super();
+  heartbeatIncoming: 30000,
-    this.options = options;
+  heartbeatOutgoing: 30000,
-    this.cli = new StompClient({
+};
-      brokerURL: options.wsUrl,
+
+export class StompCli {
+  private static client: StompClient;
+  private static options: StompCliOption;
+  private static appMsgBroker: AppWsMsgBroker[] = [];
+  /**
+   * key-订阅路径
+   */
+  subscriptions: Map<string, AppStateSubscription> = new Map<
+    string,
+    AppStateSubscription
+  >();
+  private static connected = false;
+  static new(options: StompCliOption) {
+    if (StompCli.client) {
+      // 已经创建
+      return;
+    }
+    StompCli.options = Object.assign({}, DefaultStompOption, options);
+    StompCli.client = new StompClient({
+      brokerURL: StompCli.options.wsUrl,
       connectHeaders: {
-        Authorization: options.token ? options.token : '',
+        Authorization: StompCli.options.token ? StompCli.options.token : '',
       },
-      reconnectDelay: ReconnectDelay,
+      reconnectDelay: StompCli.options.reconnectDelay,
-      heartbeatIncoming: HeartbeatIncoming,
+      heartbeatIncoming: StompCli.options.heartbeatIncoming,
-      heartbeatOutgoing: HeartbeatOutgoing,
+      heartbeatOutgoing: StompCli.options.heartbeatOutgoing,
     });
 
-    this.cli.onConnect = () => {
+    StompCli.client.onConnect = () => {
-      this.emit('connected', '');
+      // console.log('websocket连接(重连),重新订阅', StompCli.appMsgBroker.length)
+      StompCli.emitConnectStateChangeEvent(true);
+      StompCli.appMsgBroker.forEach((broker) => {
+        broker.resubscribe();
+      });
     };
-    this.cli.onStompError = (frame: Frame) => {
+
+    StompCli.client.onStompError = (frame: Frame) => {
       const errMsg = frame.headers['message'];
       if (errMsg === '401') {
         console.warn('认证失败,断开WebSocket连接');
-        this.cli.deactivate();
+        StompCli.close();
+        if (StompCli.options.onAuthenticationFailed) {
+          StompCli.options.onAuthenticationFailed();
+        }
       } else {
         console.error('收到Stomp错误消息', frame);
       }
     };
 
-    this.cli.onDisconnect = (frame: Frame) => {
+    StompCli.client.onDisconnect = (frame: Frame) => {
       console.log('Stomp 断开连接', frame);
-      this.emit('disconnected', frame);
+      StompCli.emitConnectStateChangeEvent(false);
+    };
+    StompCli.client.onWebSocketClose = (evt: CloseEvent) => {
+      console.log('websocket 关闭', evt);
+      StompCli.emitConnectStateChangeEvent(false);
     };
     // websocket错误处理
-    this.cli.onWebSocketError = (err: Event) => {
+    StompCli.client.onWebSocketError = (err: Event) => {
-      console.error('websocket错误', err);
+      console.log('websocket错误', err);
     };
 
-    this.cli.activate();
+    StompCli.client.activate();
   }
 
-  get connected(): boolean {
+  static emitConnectStateChangeEvent(connected: boolean) {
-    return this.cli.connected;
+    StompCli.connected = connected;
+    StompCli.appMsgBroker.forEach((broker) => {
+      broker.app.emit('websocket-connect-state-change', connected);
+    });
   }
 
-  subscribe(destination: string, handle: MessageHandler): ISubscription {
+  static isConnected(): boolean {
-    const sub = this.cli.subscribe(
+    return StompCli.client && StompCli.client.connected;
-      destination,
+  }
-      (frame) => {
+
-        if (this.options.protocol === 'json') {
+  static trySubscribe(
-          const data = JSON.parse(frame.body);
+    destination: string,
-          handle(data);
+    handler: messageCallbackType
-        } else {
+  ): StompSubscription | undefined {
-          handle(frame.binaryBody);
+    if (StompCli.isConnected()) {
-        }
+      return StompCli.client.subscribe(destination, handler, {
-      },
-      {
         id: destination,
-      }
+      });
-    );
+    }
-    return sub;
+    return undefined;
   }
 
-  close(): void {
+  static registerAppMsgBroker(broker: AppWsMsgBroker) {
-    this.cli.deactivate();
+    StompCli.appMsgBroker.push(broker);
+  }
+
+  static removeAppMsgBroker(broker: AppWsMsgBroker) {
+    const index = StompCli.appMsgBroker.findIndex((mb) => mb == broker);
+    if (index >= 0) {
+      StompCli.appMsgBroker.splice(index, 1);
+    }
+  }
+
+  static hasAppMsgBroker(): boolean {
+    return StompCli.appMsgBroker.length > 0;
+  }
+
+  /**
+   * 关闭websocket连接
+   */
+  static close() {
+    StompCli.connected = false;
+    if (StompCli.client) {
+      StompCli.client.deactivate();
+    }
+  }
+}
+
+// 状态订阅消息转换器
+export type GraphicStateMessageConvert = (
+  message: Uint8Array
+) => GraphicState[];
+
+// 订阅消息处理器
+export type SubscriptionMessageHandle = (message: Uint8Array) => void;
+
+// 图形app状态订阅
+export interface AppStateSubscription {
+  /**
+   * 订阅路径
+   */
+  destination: string;
+  /**
+   * 图形状态消息转换
+   */
+  messageConverter?: GraphicStateMessageConvert;
+  /**
+   * 订阅消息处理
+   */
+  messageHandle?: SubscriptionMessageHandle;
+  /**
+   * 订阅成功对象，用于取消订阅
+   * 非客户端使用
+   */
+  subscription?: StompSubscription;
+}
+
+/**
+ * 图形APP的websocket消息代理
+ */
+export class AppWsMsgBroker {
+  app: GraphicApp;
+  subscriptions: Map<string, AppStateSubscription> = new Map<
+    string,
+    AppStateSubscription
+  >();
+
+  constructor(app: GraphicApp) {
+    this.app = app;
+    StompCli.registerAppMsgBroker(this);
+  }
+
+  subscribe(sub: AppStateSubscription) {
+    this.unsbuscribe(sub.destination); // 先尝试取消之前订阅
+    sub.subscription = StompCli.trySubscribe(
+      sub.destination,
+      (message: Message) => {
+        if (sub.messageConverter) {
+          const graphicStates = sub.messageConverter(message.binaryBody);
+          this.app.handleGraphicStates(graphicStates);
+        } else if (sub.messageHandle) {
+          sub.messageHandle(message.binaryBody);
+        } else {
+          console.error(
+            `订阅destination:${sub.destination}没有消息处理器或图形状态消息转换器`
+          );
+        }
+      }
+    );
+    // console.log('代理订阅结果', sub.subscription)
+    this.subscriptions.set(sub.destination, sub);
+  }
+
+  resubscribe() {
+    this.subscriptions.forEach((record) => {
+      this.subscribe(record);
+    });
+  }
+
+  unsbuscribe(destination: string) {
+    const oldSub = this.subscriptions.get(destination);
+    if (oldSub) {
+      if (oldSub.subscription && StompCli.isConnected()) {
+        oldSub.subscription.unsubscribe();
+      }
+      oldSub.subscription = undefined;
+    }
+  }
+
+  unsbuscribeAll() {
+    this.subscriptions.forEach((record) => {
+      this.unsbuscribe(record.destination);
+    });
+  }
+
+  /**
+   * 取消所有订阅，从通用Stomp客户端移除此消息代理
+   */
+  close() {
+    StompCli.removeAppMsgBroker(this);
+    this.unsbuscribeAll();
   }
 }

src/jl-graphic/message/index.ts

@@ -1 +1 @@
-export * from './MessageBroker';
+export * from './WsMsgBroker';

yarn.lock

@@ -195,7 +195,7 @@
 
 "@pixi/graphics-extras@^7.2.4":
   version "7.2.4"
-  resolved "https://registry.npmjs.org/@pixi/graphics-extras/-/graphics-extras-7.2.4.tgz"
+  resolved "https://registry.yarnpkg.com/@pixi/graphics-extras/-/graphics-extras-7.2.4.tgz#72ac967992f239d3671d6e680ac891471619fe07"
   integrity sha512-0yT91yqF3KLiZI/iLRcfcYlTVpkVyWsfGtWEIorZs0eX+/zYx7um7EJ2h7tFORI/1FxA2maR4td5vpgCwOLJAQ==
 
 "@pixi/graphics@7.2.4":
@@ -314,59 +314,6 @@
     eventemitter3 "^4.0.0"
     url "^0.11.0"
 
-"@protobufjs/aspromise@^1.1.1", "@protobufjs/aspromise@^1.1.2":
-  version "1.1.2"
-  resolved "https://registry.npmmirror.com/@protobufjs/aspromise/-/aspromise-1.1.2.tgz"
-  integrity sha512-j+gKExEuLmKwvz3OgROXtrJ2UG2x8Ch2YZUxahh+s1F2HZ+wAceUNLkvy6zKCPVRkU++ZWQrdxsUeQXmcg4uoQ==
-
-"@protobufjs/base64@^1.1.2":
-  version "1.1.2"
-  resolved "https://registry.npmmirror.com/@protobufjs/base64/-/base64-1.1.2.tgz"
-  integrity sha512-AZkcAA5vnN/v4PDqKyMR5lx7hZttPDgClv83E//FMNhR2TMcLUhfRUBHCmSl0oi9zMgDDqRUJkSxO3wm85+XLg==
-
-"@protobufjs/codegen@^2.0.4":
-  version "2.0.4"
-  resolved "https://registry.npmmirror.com/@protobufjs/codegen/-/codegen-2.0.4.tgz"
-  integrity sha512-YyFaikqM5sH0ziFZCN3xDC7zeGaB/d0IUb9CATugHWbd1FRFwWwt4ld4OYMPWu5a3Xe01mGAULCdqhMlPl29Jg==
-
-"@protobufjs/eventemitter@^1.1.0":
-  version "1.1.0"
-  resolved "https://registry.npmmirror.com/@protobufjs/eventemitter/-/eventemitter-1.1.0.tgz"
-  integrity sha512-j9ednRT81vYJ9OfVuXG6ERSTdEL1xVsNgqpkxMsbIabzSo3goCjDIveeGv5d03om39ML71RdmrGNjG5SReBP/Q==
-
-"@protobufjs/fetch@^1.1.0":
-  version "1.1.0"
-  resolved "https://registry.npmmirror.com/@protobufjs/fetch/-/fetch-1.1.0.tgz"
-  integrity sha512-lljVXpqXebpsijW71PZaCYeIcE5on1w5DlQy5WH6GLbFryLUrBD4932W/E2BSpfRJWseIL4v/KPgBFxDOIdKpQ==
-  dependencies:
-    "@protobufjs/aspromise" "^1.1.1"
-    "@protobufjs/inquire" "^1.1.0"
-
-"@protobufjs/float@^1.0.2":
-  version "1.0.2"
-  resolved "https://registry.npmmirror.com/@protobufjs/float/-/float-1.0.2.tgz"
-  integrity sha512-Ddb+kVXlXst9d+R9PfTIxh1EdNkgoRe5tOX6t01f1lYWOvJnSPDBlG241QLzcyPdoNTsblLUdujGSE4RzrTZGQ==
-
-"@protobufjs/inquire@^1.1.0":
-  version "1.1.0"
-  resolved "https://registry.npmmirror.com/@protobufjs/inquire/-/inquire-1.1.0.tgz"
-  integrity sha512-kdSefcPdruJiFMVSbn801t4vFK7KB/5gd2fYvrxhuJYg8ILrmn9SKSX2tZdV6V+ksulWqS7aXjBcRXl3wHoD9Q==
-
-"@protobufjs/path@^1.1.2":
-  version "1.1.2"
-  resolved "https://registry.npmmirror.com/@protobufjs/path/-/path-1.1.2.tgz"
-  integrity sha512-6JOcJ5Tm08dOHAbdR3GrvP+yUUfkjG5ePsHYczMFLq3ZmMkAD98cDgcT2iA1lJ9NVwFd4tH/iSSoe44YWkltEA==
-
-"@protobufjs/pool@^1.1.0":
-  version "1.1.0"
-  resolved "https://registry.npmmirror.com/@protobufjs/pool/-/pool-1.1.0.tgz"
-  integrity sha512-0kELaGSIDBKvcgS4zkjz1PeddatrjYcmMWOlAuAPwAeccUrPHdUqo/J6LiymHHEiJT5NrF1UVwxY14f+fy4WQw==
-
-"@protobufjs/utf8@^1.1.0":
-  version "1.1.0"
-  resolved "https://registry.npmmirror.com/@protobufjs/utf8/-/utf8-1.1.0.tgz"
-  integrity sha512-Vvn3zZrhQZkkBE8LSuW3em98c0FwgO4nxzv6OdSxPKJIEKY2bGbHn+mhGIPerzI4twdxaP8/0+06HBpwf345Lw==
-
 "@quasar/app-vite@^1.0.0":
   version "1.4.3"
   resolved "https://registry.npmmirror.com/@quasar/app-vite/-/app-vite-1.4.3.tgz"
@@ -544,11 +491,6 @@
   resolved "https://registry.npmmirror.com/@types/node/-/node-20.2.5.tgz"
   integrity sha512-JJulVEQXmiY9Px5axXHeYGLSjhkZEnD+MDPDGbCbIAbMslkKwmygtZFy1X6s/075Yo94sf8GuSlFfPzysQrWZQ==
 
-"@types/node@>=13.7.0":
-  version "20.4.5"
-  resolved "https://registry.npmmirror.com/@types/node/-/node-20.4.5.tgz"
-  integrity sha512-rt40Nk13II9JwQBdeYqmbn2Q6IVTA5uPhvSO+JVqdXw/6/4glI6oR9ezty/A9Hg5u7JH4OmYmuQ+XvjKm0Datg==
-
 "@types/node@^12.20.21":
   version "12.20.55"
   resolved "https://registry.npmmirror.com/@types/node/-/node-12.20.55.tgz"
@@ -1047,14 +989,6 @@ caniuse-lite@^1.0.30001464, caniuse-lite@^1.0.30001489:
   resolved "https://registry.npmmirror.com/caniuse-lite/-/caniuse-lite-1.0.30001489.tgz"
   integrity sha512-x1mgZEXK8jHIfAxm+xgdpHpk50IN3z3q3zP261/WS+uvePxW8izXuCu6AHz0lkuYTlATDehiZ/tNyYBdSQsOUQ==
 
-centrifuge@^4.0.1:
-  version "4.0.1"
-  resolved "https://registry.npmmirror.com/centrifuge/-/centrifuge-4.0.1.tgz"
-  integrity sha512-akAyUfvMnyoCa6X2tdU5WPEccVqjkb9R/xxfm3iJ7ha7EIawjKu1dQmoyMkoLDpgl0jFxhorjmVgRcbojE17DA==
-  dependencies:
-    events "^3.3.0"
-    protobufjs "^7.2.4"
-
 chalk@^4.0.0, chalk@^4.1.0, chalk@^4.1.1:
   version "4.1.2"
   resolved "https://registry.npmmirror.com/chalk/-/chalk-4.1.2.tgz"
@@ -1774,11 +1708,6 @@ eventemitter3@^4.0.0:
   resolved "https://registry.npmmirror.com/eventemitter3/-/eventemitter3-4.0.7.tgz"
   integrity sha512-8guHBZCwKnFhYdHr2ysuRWErTwhoN2X8XELRlrRwpmfeY2jjuUN4taQMsULKUVo1K4DvZl+0pgfyoysHxvmvEw==
 
-events@^3.3.0:
-  version "3.3.0"
-  resolved "https://registry.npmmirror.com/events/-/events-3.3.0.tgz"
-  integrity sha512-mQw+2fkQbALzQ7V0MY0IqdnXNOeTtP4r0lN9z7AAawCXgqea7bDii20AYrIBrFd/Hx0M2Ocz6S111CaFkUcb0Q==
-
 express@^4.17.3:
   version "4.18.2"
   resolved "https://registry.npmmirror.com/express/-/express-4.18.2.tgz"
@@ -2387,11 +2316,6 @@ log-symbols@^4.1.0:
     chalk "^4.1.0"
     is-unicode-supported "^0.1.0"
 
-long@^5.0.0:
-  version "5.2.3"
-  resolved "https://registry.npmmirror.com/long/-/long-5.2.3.tgz"
-  integrity sha512-lcHwpNoggQTObv5apGNCTdJrO69eHOZMi4BNC+rTLER8iHAqGrUVeLh/irVIM7zTw2bOXA8T6uNPeujwOLg/2Q==
-
 lower-case@^1.1.1:
   version "1.1.4"
   resolved "https://registry.npmmirror.com/lower-case/-/lower-case-1.1.4.tgz"
@@ -2780,24 +2704,6 @@ process-nextick-args@~2.0.0:
   resolved "https://registry.npmmirror.com/process-nextick-args/-/process-nextick-args-2.0.1.tgz"
   integrity sha512-3ouUOpQhtgrbOa17J7+uxOTpITYWaGP7/AhoR3+A+/1e9skrzelGi/dXzEYyvbxubEF6Wn2ypscTKiKJFFn1ag==
 
-protobufjs@^7.2.4:
-  version "7.2.4"
-  resolved "https://registry.npmmirror.com/protobufjs/-/protobufjs-7.2.4.tgz"
-  integrity sha512-AT+RJgD2sH8phPmCf7OUZR8xGdcJRga4+1cOaXJ64hvcSkVhNcRHOwIxUatPH15+nj59WAGTDv3LSGZPEQbJaQ==
-  dependencies:
-    "@protobufjs/aspromise" "^1.1.2"
-    "@protobufjs/base64" "^1.1.2"
-    "@protobufjs/codegen" "^2.0.4"
-    "@protobufjs/eventemitter" "^1.1.0"
-    "@protobufjs/fetch" "^1.1.0"
-    "@protobufjs/float" "^1.0.2"
-    "@protobufjs/inquire" "^1.1.0"
-    "@protobufjs/path" "^1.1.2"
-    "@protobufjs/pool" "^1.1.0"
-    "@protobufjs/utf8" "^1.1.0"
-    "@types/node" ">=13.7.0"
-    long "^5.0.0"
-
 protoc-gen-ts@^0.8.6:
   version "0.8.6"
   resolved "https://registry.npmmirror.com/protoc-gen-ts/-/protoc-gen-ts-0.8.6.tgz"