Class MessageQueueGroup

packages/message/message_queue.ts:97–170

// 消息队列分组
export class MessageQueueGroup implements IMessageQueue {
  private middlewares: MiddlewareFunction[] = [];

  constructor(
    private messageQueue: IMessageQueue,
    private name: string,
    middleware?: MiddlewareFunction
  ) {
    if (!name.endsWith("/") && name.length > 0) {
      this.name += "/";
    }
    if (middleware) {
      this.middlewares.push(middleware);
    }
  }

  // 创建子分组
  group(name: string, middleware?: MiddlewareFunction) {
    const newGroup = new MessageQueueGroup(this.messageQueue, `${this.name}${name}`, middleware);
    // 继承父级的中间件
    newGroup.middlewares = [...this.middlewares, ...newGroup.middlewares];
    return newGroup;
  }

  // 添加中间件
  use(middleware: MiddlewareFunction) {
    this.middlewares.push(middleware);
    return this;
  }

  // 订阅消息
  subscribe<T>(topic: string, handler: MessageHandler<T>) {
    const fullTopic = `${this.name}${topic}`;

    if (this.middlewares.length === 0) {
      // 没有中间件，直接订阅
      return this.messageQueue.subscribe(fullTopic, handler);
    } else {
      // 有中间件，需要包装处理函数
      const wrappedHandler = async (message: T) => {
        let index = 0;

        const next = async (): Promise<void> => {
          if (index < this.middlewares.length) {
            const middleware = this.middlewares[index++];
            const result = middleware(fullTopic, message, next);
            if (result instanceof Promise) {
              await result;
            }
          } else {
            // 所有中间件都执行完毕，执行最终的处理函数
            handler(message);
          }
        };

        await next();
      };