@eduardorothdev/rxjs-mqtt

RxJS Wrapper for mqtt with TS support. Based on async-mqtt

Install

npm i @eduardorothdev/rxjs-mqtt

Using it

You first need to call and await the connect function, which will return a MqttClient object with updated methods.

The methods for this MqttClient object are the same as one generated with mqtt but has the following methods modified:

• Promises
  • .publish()
  • .subscribe()
  • .unsubscribe()
  • .end()
• Observables
  • .on()
    • This method returns an observable that emits any time it receives from the event being listened.
  • .onJsonMessage<T>() New method
    • This is a helper method that extends the on() one, getting only the events of message sent from the MQTT server, and parsing them to JSON
    • <T> is the interface or type that the method will try to parse the received message

In addition to those, I added two more Operator Function helpers that you can chain to the pipe method of the Observable returned, that will help you easily parse the received events.

• Helpers
  • parsePayload<T>(parser: (buffer: Buffer) => T | string = (buffer: Buffer) => buffer.toString()
    • This Operator Function helper allows you to parse the bytes payload received from the MQTT event with a custom function that will convert the bytes to anything.
  • parsePayloadToJSON()
    • This Operator Function helper extends the previous function to easily convert the received bytes to JSON.

Example code

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

License

Licensed under MIT.