Use ts strict checking for binding mqtt

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

@@ -41,6 +41,14 @@
 const { info, debug, error, warn } = createLoggers("binding-mqtt", "mqtt-broker-server");
 
 export default class MqttBrokerServer implements ProtocolServer {
+    private static brokerIsInitialized(broker?: mqtt.MqttClient): asserts broker is mqtt.MqttClient {
+        if (broker === undefined) {
+            throw new Error(
+                `Broker not initialized. You need to start the ${MqttBrokerServer.name} before you can expose things.`
+            );
+        }
+    }
+
     readonly scheme: string = "mqtt";
 
     private readonly ACTION_SEGMENT_LENGTH = 3;
@@ -51,30 +59,33 @@
     private readonly INTERACTION_NAME_SEGMENT_INDEX = 2;
     private readonly INTERACTION_EXT_SEGMENT_INDEX = 3;
 
+    private readonly defaults: MqttBrokerServerConfig = { uri: "mqtt://localhost:1883" };
+
     private port = -1;
-    private address: string = undefined;
+    private address?: string = undefined;
 
-    private brokerURI: string = undefined;
+    private brokerURI: string;
 
     private readonly things: Map<string, ExposedThing> = new Map();
 
     private readonly config: MqttBrokerServerConfig;
 
-    private broker: mqtt.MqttClient;
+    private broker?: mqtt.MqttClient;
 
-    private hostedServer: Aedes;
-    private hostedBroker: net.Server;
+    private hostedServer?: Aedes;
+    private hostedBroker?: net.Server;
 
     constructor(config: MqttBrokerServerConfig) {
-        this.config = config ?? { uri: "mqtt://localhost:1883" };
+        this.config = config ?? this.defaults;
+        this.config.uri = this.config.uri ?? this.defaults.uri;
 
-        if (config.uri !== undefined) {
-            // if there is a MQTT protocol indicator missing, add this
-            if (config.uri.indexOf("://") === -1) {
-                config.uri = this.scheme + "://" + config.uri;
-            }
-            this.brokerURI = config.uri;
+        // if there is a MQTT protocol indicator missing, add this
+        if (config.uri.indexOf("://") === -1) {
+            config.uri = this.scheme + "://" + config.uri;
         }
+
+        this.brokerURI = config.uri;
+
         if (config.selfHost) {
             this.hostedServer = Server({});
             let server;
@@ -86,7 +97,7 @@
             const parsed = new url.URL(this.brokerURI);
             const port = parseInt(parsed.port);
             this.port = port > 0 ? port : 1883;
-            this.hostedBroker = server.listen(port);
+            this.hostedBroker = server.listen(port, parsed.hostname);
             this.hostedServer.authenticate = this.selfHostAuthentication.bind(this);
         }
     }
@@ -130,6 +141,7 @@
     }
 
     private exposeProperty(name: string, propertyName: string, thing: ExposedThing) {
+        MqttBrokerServer.brokerIsInitialized(this.broker);
         const topic = encodeURIComponent(name) + "/properties/" + encodeURIComponent(propertyName);
         const property = thing.properties[propertyName];
 
@@ -143,6 +155,12 @@
             const observeListener = async (content: Content) => {
                 debug(`MqttBrokerServer at ${this.brokerURI} publishing to Property topic '${propertyName}' `);
                 const buffer = await content.toBuffer();
+
+                if (this.broker === undefined) {
+                    warn(`MqttBrokerServer at ${this.brokerURI} has no client to publish to. Probably it was closed.`);
+                    return;
+                }
+
                 this.broker.publish(topic, buffer);
             };
             thing.handleObserveProperty(propertyName, observeListener, { formIndex: property.forms.length - 1 });
@@ -158,6 +176,8 @@
     }
 
     private exposeAction(name: string, actionName: string, thing: ExposedThing) {
+        MqttBrokerServer.brokerIsInitialized(this.broker);
+
         const topic = encodeURIComponent(name) + "/actions/" + encodeURIComponent(actionName);
         this.broker.subscribe(topic);
 
@@ -179,6 +199,11 @@
         debug(`MqttBrokerServer at ${this.brokerURI} assigns '${href}' to Event '${eventName}'`);
 
         const eventListener = async (content: Content) => {
+            if (this.broker === undefined) {
+                warn(`MqttBrokerServer at ${this.brokerURI} has no client to publish to. Probably it was closed.`);
+                return;
+            }
+
             if (!content) {
                 warn(`MqttBrokerServer on port ${this.getPort()} cannot process data for Event ${eventName}`);
                 thing.handleUnsubscribeEvent(eventName, eventListener, { formIndex: event.forms.length - 1 });
@@ -199,6 +224,9 @@
             payload = rawPayload;
         } else if (typeof rawPayload === "string") {
             payload = Buffer.from(rawPayload);
+        } else {
+            warn(`MqttBrokerServer on port ${this.getPort()} received unexpected payload type`);
+            return;
         }
 
         if (segments.length === this.ACTION_SEGMENT_LENGTH) {
@@ -308,7 +336,7 @@
                 error(
                     `MqttBrokerServer at ${this.brokerURI} got error on writing to property '${
                         segments[this.INTERACTION_NAME_SEGMENT_INDEX]
-                    }': ${err.message}`
+                    }': ${err}`
                 );
             }
         } else {
@@ -322,14 +350,16 @@
 
     public async destroy(thingId: string): Promise<boolean> {
         debug(`MqttBrokerServer on port ${this.getPort()} destroying thingId '${thingId}'`);
-        let removedThing: ExposedThing;
+        let removedThing: ExposedThing | undefined;
+
         for (const name of Array.from(this.things.keys())) {
             const expThing = this.things.get(name);
             if (expThing != null && expThing.id != null && expThing.id === thingId) {
                 this.things.delete(name);
                 removedThing = expThing;
             }
         }
+
         if (removedThing) {
             info(`MqttBrokerServer succesfully destroyed '${removedThing.title}'`);
         } else {
@@ -385,16 +415,24 @@
         }
 
         if (this.hostedBroker !== undefined) {
-            await new Promise<void>((resolve) => this.hostedServer.close(() => resolve()));
-            await new Promise<void>((resolve) => this.hostedBroker.close(() => resolve()));
+            // When the broker is hosted, we need to close it.
+            // Both this.hostedBroker and this.hostedServer are defined at the same time.
+            // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+            await new Promise<void>((resolve) => this.hostedServer!.close(() => resolve()));
+            // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+            await new Promise<void>((resolve) => this.hostedBroker!.close(() => resolve()));
         }
     }
 
     public getPort(): number {
         return this.port;
     }
 
-    public getAddress(): string {
+    /**
+     *
+     * @returns the address of the broker or undefined if the Server is not started.
+     */
+    public getAddress(): string | undefined {
         return this.address;
     }
 
@@ -408,15 +446,15 @@
             for (let i = 0; i < this.config.selfHostAuthentication.length; i++) {
                 if (
                     username === this.config.selfHostAuthentication[i].username &&
-                    password.equals(Buffer.from(this.config.selfHostAuthentication[i].password))
+                    password.equals(Buffer.from(this.config.selfHostAuthentication[i].password ?? ""))
                 ) {
-                    done(undefined, true);
+                    done(null, true);
                     return;
                 }
             }
-            done(undefined, false);
+            done(null, false);
             return;
         }
-        done(undefined, true);
+        done(null, true);
     }
 }

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

@@ -17,7 +17,7 @@
  * Protocol test suite to test protocol implementations
  */
 
-import { ProtocolClient, Content, DefaultContent, createLoggers } from "@node-wot/core";
+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";
@@ -40,7 +40,7 @@
         this.scheme = "mqtt" + (secure ? "s" : "");
     }
 
-    private client: mqtt.MqttClient = undefined;
+    private client?: mqtt.MqttClient;
 
     public subscribeResource(
         form: MqttForm,
@@ -50,7 +50,7 @@
     ): Promise<Subscription> {
         return new Promise<Subscription>((resolve, reject) => {
             // get MQTT-based metadata
-            const contentType = form.contentType;
+            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;
@@ -63,15 +63,29 @@
                 this.client.subscribe(topic);
                 resolve(
                     new Subscription(() => {
+                        if (!this.client) {
+                            warn(
+                                `MQTT Client is undefined. This means that the client either failed to connect or was never initialized.`
+                            );
+                            return;
+                        }
                         this.client.unsubscribe(topic);
                     })
                 );
             }
 
             this.client.on("connect", () => {
-                this.client.subscribe(topic);
+                // In this case, the client is definitely defined.
+                // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+                this.client!.subscribe(topic);
                 resolve(
                     new Subscription(() => {
+                        if (!this.client) {
+                            warn(
+                                `MQTT Client is undefined. This means that the client either failed to connect or was never initialized.`
+                            );
+                            return;
+                        }
                         this.client.unsubscribe(topic);
                     })
                 );
@@ -166,8 +180,13 @@
         const security: TD.SecurityScheme = metadata[0];
 
         if (security.scheme === "basic") {
-            this.config.username = credentials.username;
-            this.config.password = credentials.password;
+            if (credentials === undefined) {
+                // FIXME: This error message should be reworded and adapt to logging convention
+                throw new Error("binding-mqtt: security wants to be basic but you have provided no credentials");
+            } else {
+                this.config.username = credentials.username;
+                this.config.password = credentials.password;
+            }
         }
         return true;
     }

packages/binding-mqtt/src/mqtt.ts

@@ -47,7 +47,7 @@ export class MqttForm extends Form {
 }
 
 export interface MqttClientConfig {
-    // username & password are redundated here (also find them in MqttClientSecurityParameters)
+    // username & password are redundant here (also find them in MqttClientSecurityParameters)
     // because MqttClient.setSecurity() method can inject authentication credentials into this interface
     // which will be then passed to mqtt.connect() once for all
     username?: string;

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

@@ -76,6 +76,10 @@ describe("MQTT client implementation", () => {
                                 if (!eventReceived) {
                                     eventReceived = true;
                                 } else {
+                                    if (!x.data) {
+                                        done(new Error("No data received"));
+                                        return;
+                                    }
                                     ProtocolHelpers.readStreamFully(ProtocolHelpers.toNodeStream(x.data)).then(
                                         (received) => {
                                             expect(JSON.parse(received.toString())).to.equal(++check);
@@ -134,6 +138,10 @@ describe("MQTT client implementation", () => {
                                 if (!eventReceived) {
                                     eventReceived = true;
                                 } else {
+                                    if (!x.data) {
+                                        done(new Error("No data received"));
+                                        return;
+                                    }
                                     ProtocolHelpers.readStreamFully(ProtocolHelpers.toNodeStream(x.data)).then(
                                         (received) => {
                                             expect(JSON.parse(received.toString())).to.equal(++check);

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

@@ -120,10 +120,10 @@ describe("MQTT client implementation", () => {
         beforeEach(() => {
             aedes.authenticate = function (_client, username: Readonly<string>, password: Readonly<Buffer>, done) {
                 if (username !== undefined) {
-                    done(undefined, username === "user" && password.equals(Buffer.from("pass")));
+                    done(null, username === "user" && password.equals(Buffer.from("pass")));
                     return;
                 }
-                done(undefined, true);
+                done(null, true);
             };
             const server = net.createServer(aedes.handle);
             hostedBroker = server.listen(brokerPort);

packages/binding-mqtt/tsconfig.json

@@ -2,7 +2,8 @@
     "extends": "../../tsconfig.json",
     "compilerOptions": {
         "outDir": "dist",
-        "rootDir": "src"
+        "rootDir": "src",
+        "strict": true
     },
     "include": ["src/**/*"],
     "references": [{ "path": "../td-tools" }, { "path": "../core" }]