fix(binding-mqtt): generate qos for events

Author: relu91

packages/binding-file/test/test2.txt

@@ -0,0 +1 @@
+Lorem ipsum dolor sit amet.

packages/binding-mqtt/src/mqtt-broker-server.ts

@@ -24,7 +24,7 @@ import { AuthenticateError, Client, Server, Aedes } from "aedes";
 import * as net from "net";
 import * as tls from "tls";
 import * as TD from "@node-wot/td-tools";
-import { MqttBrokerServerConfig } from "./mqtt";
+import { MqttBrokerServerConfig, MqttForm } from "./mqtt";
 import {
     ProtocolServer,
     Servient,
@@ -37,6 +37,7 @@ import {
 import { InteractionOptions } from "wot-typescript-definitions";
 import { ActionElement, PropertyElement } from "wot-thing-description-types";
 import { Readable } from "stream";
+import { mapQoS } from "./util";
 
 const { info, debug, error, warn } = createLoggers("binding-mqtt", "mqtt-broker-server");
 
@@ -196,7 +197,8 @@ export default class MqttBrokerServer implements ProtocolServer {
         const event = thing.events[eventName];
 
         const href = this.brokerURI + "/" + topic;
-        const form = new TD.Form(href, ContentSerdes.DEFAULT);
+        const form = new MqttForm(href, ContentSerdes.DEFAULT);
+        form["mqv:qos"] = "2";
         form.op = ["subscribeevent", "unsubscribeevent"];
         event.forms.push(form);
         debug(`MqttBrokerServer at ${this.brokerURI} assigns '${href}' to Event '${eventName}'`);
@@ -214,7 +216,7 @@ export default class MqttBrokerServer implements ProtocolServer {
             }
             debug(`MqttBrokerServer at ${this.brokerURI} publishing to Event topic '${eventName}' `);
             const buffer = await content.toBuffer();
-            this.broker.publish(topic, buffer);
+            this.broker.publish(topic, buffer, { retain: form["mqv:retain"], qos: mapQoS(form["mqv:qos"]) });
         };
         thing.handleSubscribeEvent(eventName, eventListener, { formIndex: event.forms.length - 1 });
     }

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

@@ -20,12 +20,12 @@
 import { ProtocolClient, Content, DefaultContent, createLoggers, ContentSerdes } from "@node-wot/core";
 import * as TD from "@node-wot/td-tools";
 import * as mqtt from "mqtt";
-import { MqttClientConfig, MqttForm, MqttQoS } from "./mqtt";
+import { MqttClientConfig, MqttForm } from "./mqtt";
 import * as url from "url";
 import { Subscription } from "rxjs/Subscription";
 import { Readable } from "stream";
-import { IClientPublishOptions } from "mqtt";
 import MQTTMessagePool from "./mqtt-message-pool";
+import { mapQoS } from "./util";
 
 const { debug, warn } = createLoggers("binding-mqtt", "mqtt-client");
 
@@ -130,7 +130,7 @@ export default class MqttClient implements ProtocolClient {
         const buffer = content === undefined ? Buffer.from("") : await content.toBuffer();
         await pool.publish(topic, buffer, {
             retain: form["mqv:retain"],
-            qos: this.mapQoS(form["mqv:qos"]),
+            qos: mapQoS(form["mqv:qos"]),
         });
     }
 
@@ -152,7 +152,7 @@ export default class MqttClient implements ProtocolClient {
         const buffer = content === undefined ? Buffer.from("") : await content.toBuffer();
         await pool.publish(topic, buffer, {
             retain: form["mqv:retain"],
-            qos: this.mapQoS(form["mqv:qos"]),
+            qos: mapQoS(form["mqv:qos"]),
         });
         // there will be no response
         return new DefaultContent(Readable.from([]));
@@ -206,21 +206,4 @@ export default class MqttClient implements ProtocolClient {
         }
         return true;
     }
-
-    private mapQoS(qos: MqttQoS | undefined): Required<IClientPublishOptions>["qos"] {
-        switch (qos) {
-            case "0":
-                return 0;
-            case "1":
-                return 1;
-            case "2":
-                return 2;
-            case undefined:
-                return 0;
-            default:
-                warn(`MqttClient received unsupported QoS level '${qos}'`);
-                warn(`MqttClient falling back to QoS level '0'`);
-                return 0;
-        }
-    }
 }

packages/binding-mqtt/src/util.ts

@@ -0,0 +1,36 @@
+/********************************************************************************
+ * Copyright (c) 2023 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v. 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the W3C Software Notice and
+ * Document License (2015-05-13) which is available at
+ * https://www.w3.org/Consortium/Legal/2015/copyright-software-and-document.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR W3C-20150513
+ ********************************************************************************/
+import { createLoggers } from "@node-wot/core";
+import { MqttQoS } from "./mqtt";
+import { IClientPublishOptions } from "mqtt";
+
+const { debug, warn } = createLoggers("binding-mqtt", "mqtt-util");
+
+export function mapQoS(qos: MqttQoS | undefined): Required<IClientPublishOptions>["qos"] {
+    switch (qos) {
+        case "0":
+            return 0;
+        case "1":
+            return 1;
+        case "2":
+            return 2;
+        case undefined:
+            return 0;
+        default:
+            warn(`MqttClient received unsupported QoS level '${qos}'`);
+            warn(`MqttClient falling back to QoS level '0'`);
+            return 0;
+    }
+}