Method handleMessage

packages/microservices/server/server-mqtt.ts:129–167

(
    channel: string,
    buffer: Buffer,
    pub: MqttClient,
    originalPacket?: Record<string, any>,
  )

  }

  public async handleMessage(
    channel: string,
    buffer: Buffer,
    pub: MqttClient,
    originalPacket?: Record<string, any>,
  ): Promise<any> {
    const rawPacket = this.parseMessage(buffer.toString());
    const packet = await this.deserializer.deserialize(rawPacket, { channel });
    const mqttContext = new MqttContext([channel, originalPacket!]);
    if (isUndefined((packet as IncomingRequest).id)) {
      return this.handleEvent(channel, packet, mqttContext);
    }
    const publish = this.getPublisher(
      pub,
      mqttContext,
      (packet as IncomingRequest).id,
    );
    const handler = this.getHandlerByPattern(channel);

    if (!handler) {
      const status = 'error';
      const noHandlerPacket = {
        id: (packet as IncomingRequest).id,
        status,
        err: NO_MESSAGE_HANDLER,
      };
      return publish(noHandlerPacket);
    }
    return this.onProcessingStartHook(
      this.transportId,
      mqttContext,
      async () => {
        const response$ = this.transformToObservable(
          await handler(packet.data, mqttContext),
        );
        response$ && this.send(response$, publish);
      },
    );
  }

  public getPublisher(
    client: MqttClient,