NestJS MQTT 微服务的有效@MessagePattern 是什么？

Question

我正在尝试根据文档使用 NestJS 设置 MQTT 微服务。

我已经使用 Docker 启动了一个有效的 Mosquitto Broker，并使用各种 MQTT 客户端验证了它的可操作性。现在，当我启动 NestJS 服务时，它似乎连接正确（mqqt.fx 显示新客户端），但我无法在我的控制器中接收任何消息。这是我的引导，就像在文档中一样：

主文件

async function bootstrap() {
    const app = await NestFactory.createMicroservice(AppModule, {
        transport: Transport.MQTT,
        options: {
            host: 'localhost',
            port: 1883,
            protocol: 'tcp'
        }
    });
    app.listen(() => console.log('Microservice is listening'));
}
bootstrap();

app.controller.ts

@Controller()
export class AppController {

    @MessagePattern('mytopic') // tried {cmd:'mytopic'} or {topic:'mytopic'}
    root(msg: Buffer) {
        console.log('received: ', msg)
    }
}

我是否错误地使用了消息模式装饰器，或者我对 NestJS MQTT 微服务甚至应该做什么的概念有误？我认为它可能会订阅我传递给装饰器的主题。我唯一的其他信息来源是相应的单元测试

Answer 1

Nest.js 模式处理程序

在 Nest.js 方面，我们有以下模式处理程序：

@MessagePattern('sum')
sum(data: number[]): number {
  return data.reduce((a, b) => a + b, 0);
}

正如@Alexandre所解释的，这实际上会监听sum_ack.

非 Nest.js 客户端

非 Nest.js 客户端可能如下所示（只需另存为 client.js，运行npm install mqtt并使用 运行程序node client.js）：

var mqtt = require('mqtt')
var client  = mqtt.connect('mqtt://localhost:1883')

client.on('connect', function () {
  client.subscribe('sum_res', function (err) {
    if (!err) {
      client.publish('sum_ack', '{"data": [2, 3]}');
    }
  })
})

client.on('message', function (topic, message) {
  console.log(message.toString())
  client.end()
})

它发送有关该主题的消息sum_ack并侦听 上的消息sum_res。当它在 上收到消息时sum_res，它会记录该消息并结束程序。Nest.js 期望消息格式为{data: myData}，然后调用参数处理程序sum(myData)。

// Log:
{"err":null,"response":5} // This is the response from sum()
{"isDisposed":true} // Internal "complete event" (according to unit test)

当然，这样不太方便……

Nest.js 客户端

这是因为这意味着与另一个 Nest.js 客户端一起使用，而不是与普通的 mqtt 客户端一起使用。Nest.js 客户端抽象了所有内部逻辑。请参阅这个答案，它描述了 redis 的客户端（mqtt 只需要更改两行）。

async onModuleInit() {
  await this.client.connect();
  // no 'sum_ack' or {data: [0, 2, 3]} needed
  this.client.send('sum', [0, 2, 3]).toPromise();
}