fastify-rxjs-mqtt

Wrapper for rxjs-mqtt with TS support.

Install

npm i @eduardorothdev/fastify-rxjs-mqtt

Using it

Add it to your fastify project with the register method.

1import fastify from "fastify";
2import { fastifyRxjsMqtt } from "@eduardorothdev/fastify-rxjs-mqtt";
3
4const host = process.env.HOST ?? "localhost";
5const port = process.env.PORT ? Number(process.env.PORT) : 3000;
6
7const server = fastify();
8
9// register plugin
10server.register(fastifyRxjsMqtt, { url: "mqtt://localhost:1883" });
11
12server.listen({ port, host }, (err) => {
13  if (err) throw err;
14});

Then you can use it in your methods

1import { FastifyInstance } from "fastify";
2
3export default async function (fastify: FastifyInstance) {
4  fastify.get("/mqtt/ping", async function (req, reply) {
5    await this.mqttClient.publish("your/topic/#", "Async Hi Mosquitto!");
6    reply.send({ mqtt: "message sent!" });
7  });
8}

Listening for events

1import { FastifyInstance } from "fastify";
2
3export default async function (fastify: FastifyInstance) {
4  try {
5    // subscribe to a topic
6    await fastify.mqttClient.subscribe("some/topic/#");
7    const sub = fastify.mqttClient
8      .onJsonMessage<{
9        some: string;
10        property: string;
11        mapping: boolean;
12      }>()
13      .pipe(
14        catchError((err) => {
15          // we have to catch the error so the
16          // observable pipe doesn't stop sending messages
17          return of(null);
18        }),
19      )
20      .subscribe((jsonMessage) => {
21        // { some: 'hello', property: 'from mqtt', mapping: true }
22        console.log(jsonMessage);
23      });
24
25    // later you can unsubscribe when needed.
26    // this will unsubscribe from the onJsonMessage observable pipe
27    // not from the topic subscription
28    // sub.unsubscribe();
29
30    // Unsubscribe from the topic subscription
31    // await fastify.mqttClient.unsubscribe('some/topic/#');
32  } catch (err) {
33    // connection/subscription-to-topic errors
34    console.log(err);
35  }
36}

License

Licensed under MIT.