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

问题描述

我正在尝试根据文档使用NestJS设置MQTT微服务. /p>

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

main.ts

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微服务应该做的事情有误?我认为它可能会订阅我传递给装饰器的主题.我唯一的其他信息来源是相应的单元测试

解决方案

nest.js模式处理程序

在nest.js方面，我们具有以下模式处理程序:

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

如 @Alexandre 所述，这实际上将收听sum_ack.

---

Non-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();
}

I'm trying to setup a MQTT Microservice using NestJS according to the docs.

I've started a working Mosquitto Broker using Docker and verified it's operability using various MQTT clients. Now, when I start the NestJS service it seems to be connecting correctly (mqqt.fx shows new client), yet I am unable to receive any messages in my controllers.This is my bootstrapping, just like in the docs:

main.ts

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)
    }
}

Am I using the message-pattern decorator wrongly or is my concept wrong of what a NestJS MQTT microservice even is supposed to do? I thought it might subscribe to the topic I pass to the decorator. My only other source of information being the corresponding unit tests

解决方案

nest.js Pattern Handler

On nest.js side we have the following pattern handler:

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

As @Alexandre explained, this will actually listen to sum_ack.

---

Non-nest.js Client

A non-nest.js client could look like this (just save as client.js, run npm install mqtt and run the program with 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()
})

It sends a message on the topic sum_ack and listens to messages on sum_res. When it receives a message on sum_res, it logs the message and ends the program. nest.js expects the message format to be {data: myData} and then call the param handler sum(myData).

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

Of course, this is not very convenient...

---

nest.js Client

That is because this is meant to be used with another nest.js client rather than a normal mqtt client. The nest.js client abstracts all the internal logic away. See this answer, which describes the client for redis (only two lines need to be changed for mqtt).

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

这篇关于NestJS MQTT微服务有效的@MessagePattern是什么?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！