refactor(binding-mqtt): use the new vocabulary

Author: relu91

packages/binding-mqtt/src/mqtt-client.ts

@@ -50,16 +50,18 @@ export default class MqttClient implements ProtocolClient {
     ): Promise<Subscription> {
         const contentType = form.contentType ?? ContentSerdes.DEFAULT;
         const requestUri = new url.URL(form.href);
-        const topic = requestUri.pathname.slice(1);
         const brokerUri: string = `${this.scheme}://` + requestUri.host;
+        // Keeping the path as the topic for compatibility reasons.
+        // Current specification allows only form["mqv:filter"]
+        const filter = requestUri.pathname.slice(1) ?? form["mqv:filter"];
 
         if (this.client === undefined) {
             this.client = await mqtt.connectAsync(brokerUri, this.config);
         }
 
         this.client.on("message", (receivedTopic: string, payload: Buffer) => {
             debug(`Received MQTT message (topic: ${receivedTopic}, data length: ${payload.length})`);
-            if (receivedTopic === topic) {
+            if (filter.includes(receivedTopic)) {
                 next(new Content(contentType, Readable.from(payload)));
             }
         });
@@ -70,7 +72,7 @@ export default class MqttClient implements ProtocolClient {
             if (error) error(err);
         });
 
-        await this.client.subscribeAsync(topic);
+        await this.client.subscribeAsync(filter);
 
         return new Subscription(() => {
             if (!this.client) {
@@ -79,7 +81,7 @@ export default class MqttClient implements ProtocolClient {
                 );
                 return;
             }
-            this.client.unsubscribe(topic);
+            this.client.unsubscribe(filter);
         });
     }
 
@@ -89,20 +91,19 @@ export default class MqttClient implements ProtocolClient {
 
     public async writeResource(form: MqttForm, content: Content): Promise<void> {
         const requestUri = new url.URL(form.href);
-        const topic = requestUri.pathname.slice(1);
         const brokerUri = `${this.scheme}://${requestUri.host}`;
+        const topic = requestUri.pathname.slice(1) ?? form["mqv:topic"];
 
         if (this.client === undefined) {
             this.client = await mqtt.connectAsync(brokerUri, this.config);
         }
 
         // if not input was provided, set up an own body otherwise take input as body
-        if (content === undefined) {
-            await this.client.publishAsync(topic, Buffer.from(""));
-        } else {
-            const buffer = await content.toBuffer();
-            await this.client.publishAsync(topic, buffer);
-        }
+        const buffer = content === undefined ? Buffer.from("") : await content.toBuffer();
+        await this.client.publishAsync(topic, buffer, {
+            retain: form["mqv:retain"],
+            qos: this.mapQoS(form["mqv:qos"]),
+        });
     }
 
     public async invokeResource(form: MqttForm, content: Content): Promise<Content> {
@@ -115,12 +116,11 @@ export default class MqttClient implements ProtocolClient {
         }
 
         // if not input was provided, set up an own body otherwise take input as body
-        if (content === undefined) {
-            await this.client.publishAsync(topic, Buffer.from(""));
-        } else {
-            const buffer = await content.toBuffer();
-            await this.client.publishAsync(topic, buffer);
-        }
+        const buffer = content === undefined ? Buffer.from("") : await content.toBuffer();
+        await this.client.publishAsync(topic, buffer, {
+            retain: form["mqv:retain"],
+            qos: this.mapQoS(form["mqv:qos"]),
+        });
         // there will be no response
         return new DefaultContent(Readable.from([]));
     }
@@ -169,15 +169,20 @@ export default class MqttClient implements ProtocolClient {
         return true;
     }
 
-    private mapQoS(qos: MqttQoS): Required<IClientPublishOptions>["qos"] {
+    private mapQoS(qos: MqttQoS | undefined): Required<IClientPublishOptions>["qos"] {
         switch (qos) {
-            case 2:
-                return (qos = 2);
-            case 1:
-                return (qos = 1);
-            case 0:
+            case "0":
+                return 0;
+            case "1":
+                return 1;
+            case "2":
+                return 2;
+            case undefined:
+                return 0;
             default:
-                return (qos = 0);
+                warn(`MqttClient received unsupported QoS level '${qos}'`);
+                warn(`MqttClient falling back to QoS level '0'`);
+                return 0;
         }
     }
 }

packages/binding-mqtt/src/mqtt.ts

@@ -29,21 +29,16 @@ export * from "./mqtt-client-factory";
 export * from "./mqtts-client-factory";
 export * from "./mqtt-broker-server";
 
-/**
- * MQTT Quality of Service level.
- * QoS0: Fire-and-forget
- * QoS1: Deliver-at-least-once
- * QoS2: Deliver-exactly-once
- */
-export enum MqttQoS {
-    QoS0,
-    QoS1,
-    QoS2,
-}
-
+export type MqttQoS = "0" | "1" | "2";
 export class MqttForm extends Form {
-    public "mqtt:qos": MqttQoS = MqttQoS.QoS0;
-    public "mqtt:retain": boolean;
+    public "mqv:qos"?: MqttQoS = "0";
+    public "mqv:retain"?: boolean;
+
+    public "mqv:topic"?: string;
+
+    public "mqv:filter"?: string | string[];
+
+    public "mqv:controlPacket"?: "publish" | "subscribe" | "unsubscribe";
 }
 
 export interface MqttClientConfig {

packages/binding-mqtt/test/mqtt-client-subscribe-test.unit.ts

@@ -19,7 +19,7 @@
 
 import * as chai from "chai";
 import chaiAsPromised from "chai-as-promised";
-import { MqttClient, MqttForm, MqttQoS } from "../src/mqtt";
+import { MqttClient, MqttForm } from "../src/mqtt";
 import { expect, should } from "chai";
 import { Aedes, Server } from "aedes";
 import * as net from "net";
@@ -61,8 +61,8 @@ describe("MQTT client implementation - unit", () => {
             const mqttClient = new MqttClient();
             const form: MqttForm = {
                 href: brokerUri + "/" + property,
-                "mqtt:qos": MqttQoS.QoS0,
-                "mqtt:retain": false,
+                "mqv:qos": "0",
+                "mqv:retain": false,
             };
 
             mqttClient
@@ -88,8 +88,8 @@ describe("MQTT client implementation - unit", () => {
             const mqttClient = new MqttClient();
             const form: MqttForm = {
                 href: brokerUri + "/" + property,
-                "mqtt:qos": MqttQoS.QoS0,
-                "mqtt:retain": false,
+                "mqv:qos": "0",
+                "mqv:retain": false,
             };
 
             mqttClient
@@ -139,8 +139,8 @@ describe("MQTT client implementation - unit", () => {
 
             const form: MqttForm = {
                 href: brokerUri + "/" + property,
-                "mqtt:qos": MqttQoS.QoS1,
-                "mqtt:retain": false,
+                "mqv:qos": "1",
+                "mqv:retain": false,
             };
 
             mqttClient
@@ -158,8 +158,8 @@ describe("MQTT client implementation - unit", () => {
 
             const form: MqttForm = {
                 href: brokerUri + "/" + property,
-                "mqtt:qos": MqttQoS.QoS1,
-                "mqtt:retain": false,
+                "mqv:qos": "1",
+                "mqv:retain": false,
             };
 
             await mqttClient.subscribeResource(form, () => {});