Waiting for mqtt answer possible

example/src/index.ts

@@ -1,6 +1,7 @@
 import { MultiMessageBroker, FileImporter } from "aas-multimessagebroker";
 
 import HTTPInterfaceServer from "./modules/httpInterfaceServer";
+// Change to "./modules/mqttConnectorVariant" to use blocking in callActionSync
 import MQTTConnector from "./modules/mqttConnector";
 
 /**

example/src/modules/mqttConnectorVariant.ts

@@ -0,0 +1,172 @@
+import * as mqtt from "mqtt";
+import deasync from "deasync";
+import { AbstractConnectionObject } from "aas-multimessagebroker";
+import { AASCoreTypes, type Types } from "aas-multimessagebroker";
+
+/**
+ * This class is the same as the {@link MQTTConnector|MQTTConnector} but
+ * implements waiting for an answer on callActionSync()
+ *! Attention: this is hardcoded for the example asset!
+ *! Do not use in productive code!
+ * 
+ *  @see {@link MQTTConnector|MQTTConnector}
+ */
+export default class MQTTConnectorLock extends AbstractConnectionObject<mqtt.IClientOptions> {
+    public static readonly connectorName: string = "MQTT Connector";
+    public static readonly uriProtocol: string[] = ["mqtt", "mqtts"];
+    public static readonly connectionType: "ON_DEMAND" | "PERMANENT" = "PERMANENT";
+    public static readonly supportsSubscriptions: boolean = true;
+
+    private client: mqtt.MqttClient|null = null;
+    private readonly messageStore: Record<string, any> = {};
+
+    public connect(): boolean {
+        if (!this.client) this.client = mqtt.connect(this.endpointMetadata.base, this.connectionParameter);
+        this.client.on("message", (topic, message) => {
+            //console.log(`${topic}: ${message.toString("utf-8")}`)
+            // If json is expected you could parse it here
+            this.messageStore[topic] = message.toString("utf-8");
+            // Notify observers
+            if (this.observerStore[topic] !== undefined) {
+                this.observerStore[topic].forEach(sub => sub.cb({ type: "event", value: message.toString("utf-8"), target: sub.target}));
+            }
+            if (this.eventSubStore[topic] !== undefined) {
+                this.eventSubStore[topic].forEach(sub => sub.cb({ type: "event", value: message.toString("utf-8"), target: sub.target}));
+            }
+        });
+        this.client.subscribe("#");
+        return this.client.connected;
+    }
+
+    public disconnect(): void {
+        if (this.client) this.client.end();
+        this.client = null;
+    }
+
+    private assertConnected(): void {
+        if (!this.client || !this.client.connected) throw new Error("Not connected");
+    }
+
+    private static getTopicFromForm(form: Types.AIDTypes.InterfaceForm): string {
+        if (form.href.startsWith("/")) return form.href.substring(1);
+        else {
+            const url = new URL(form.href);
+            return url.pathname.substring(1);
+        }
+    }
+
+    public readProperty(_target: AASCoreTypes.Property, mapping: Types.RequestTypes.ConnectionConfiguration): any {
+        const errorReturn = mapping.default ?? null;
+        if (!this.client || !this.client.connected) return errorReturn;
+
+        const value = this.messageStore[MQTTConnectorLock.getTopicFromForm(mapping.forms[0])];
+        if (value === undefined) return errorReturn;
+        switch (mapping.type) {
+            case "integer":
+                return Number.parseInt(value);
+            case "float":
+                return Number.parseFloat(value);
+            case "boolean":
+                return !!JSON.parse(value);
+            case "string":
+            default:
+                return value;
+        }
+
+    }
+
+    public writeProperty(_target: AASCoreTypes.Property, mapping: Types.RequestTypes.ConnectionConfiguration, value: any): boolean {
+        if (!this.client || !this.client.connected) return false;
+
+        const topic = MQTTConnectorLock.getTopicFromForm(mapping.forms[0]);
+        if (topic === undefined) return false;
+
+        this.client.publish(topic, value.toString());
+
+        return true;
+    }
+
+    public observeProperty(target: AASCoreTypes.Property, mapping: Types.RequestTypes.ConnectionConfiguration, cb: (value: Types.RequestTypes.MMBEvent) => void): boolean {
+        if (!mapping.observable || !this.client || !this.client.connected) return false;
+
+        const topic = MQTTConnectorLock.getTopicFromForm(mapping.forms[0]);
+        if (this.observerStore[topic] === undefined) this.observerStore[topic] = [{ target, cb }];
+        else this.observerStore[topic].push({ target, cb });
+
+        return true;
+    }
+
+    public unobserveProperty(target: AASCoreTypes.Property, mapping: Types.RequestTypes.ConnectionConfiguration): void {
+        const topic = MQTTConnectorLock.getTopicFromForm(mapping.forms[0]);
+        this.observerStore[topic];
+        //! TODO
+    }
+
+    public callActionSync(_target: AASCoreTypes.Operation, mapping: Types.RequestTypes.ConnectionConfiguration, args: Record<string, any>): any {
+        this.assertConnected();
+
+        const form: Types.AIDTypes.InterfaceFormMQTTAction = mapping.forms[0] as Types.AIDTypes.InterfaceFormMQTTAction;
+        const topic = MQTTConnectorLock.getTopicFromForm(form);
+        if (topic === undefined || !mapping.forms) throw new Error("Mapping invalid.");
+        
+        const message = JSON.stringify(args);
+
+        
+        if (topic === "testasset/testoperation3") {
+            delete this.messageStore["testasset/testoperation3/result"];
+        }
+
+        this.execute(form.mqv_controlPacketValue ?? "PUBLISH", topic, message);
+        
+        // This is probably the worst possible approach waiting for a result
+        // Please don't do this in production but use async code or smth like that instead
+        if (topic === "testasset/testoperation3") {
+            while (this.messageStore["testasset/testoperation3/result"] === undefined) {
+                deasync.sleep(100);
+            }
+            return this.messageStore["testasset/testoperation3/result"];
+        }
+
+        return;
+    }
+
+    public callActionAsync(_target: AASCoreTypes.Operation, mapping: Types.RequestTypes.ConnectionConfiguration, args: Record<string, any>): string {
+        this.assertConnected();
+
+        const form: Types.AIDTypes.InterfaceFormMQTTAction = mapping.forms[0] as Types.AIDTypes.InterfaceFormMQTTAction;
+        const topic = MQTTConnectorLock.getTopicFromForm(form);
+        if (topic === undefined || !mapping.forms) throw new Error("Mapping invalid.");
+        
+        const message = JSON.stringify(args);
+        this.execute(form.mqv_controlPacketValue ?? "PUBLISH", topic, message);
+        
+        return this.generateAsyncHandle();
+    }
+
+    public subscribeEvent(target: AASCoreTypes.Class, mapping: Types.RequestTypes.ConnectionConfiguration,  cb: (event: Types.RequestTypes.MMBEvent) => void): boolean {
+        if (!mapping.observable || !this.client || !this.client.connected) return false;
+
+        const topic = MQTTConnectorLock.getTopicFromForm(mapping.forms[0]);
+        if (this.eventSubStore[topic] === undefined) this.eventSubStore[topic] = [{ target, cb }];
+        else this.eventSubStore[topic].push({ target, cb });
+
+        return true;
+    }
+
+    public unsubscribeEvent(event: AASCoreTypes.Class, mapping: Types.RequestTypes.ConnectionConfiguration): void {
+        // TODO
+        // How do we know which event to unsubscribe from?
+    }
+
+    private execute(packetType: "PUBLISH" | "SUBSCRIBE" | "UNSUBSCRIBE", topic: string, message?: string) {
+        this.assertConnected();
+        if (this.client){
+            switch (packetType) {
+                case "PUBLISH": return this.client.publish(topic, message ?? "1");
+                case "SUBSCRIBE": return this.client.subscribe(topic);
+                case "UNSUBSCRIBE": return this.client.unsubscribe(topic);
+            }
+        }
+    }
+
+}