diff --git a/example/package-lock.json b/example/package-lock.json index 846e8d1..4992eab 100644 --- a/example/package-lock.json +++ b/example/package-lock.json @@ -7,13 +7,15 @@ "": { "name": "aas-multimessagebroker-example", "version": "1.0.0", - "license": "ISC", + "license": "LGPL-3.0-or-later", "dependencies": { "aas-multimessagebroker": "file:../lib", + "deasync": "^0.1.29", "express": "^4.18.2", "mqtt": "^5.1.4" }, "devDependencies": { + "@types/deasync": "^0.1.5", "@types/express": "^4.17.20", "typedoc": "^0.25.3", "typedoc-plugin-extras": "^3.0.0", @@ -24,7 +26,7 @@ "../lib": { "name": "aas-multimessagebroker", "version": "1.0.0", - "license": "ISC", + "license": "LGPL-3.0-or-later", "dependencies": { "@aas-core-works/aas-core3.0-typescript": "^1.0.0-rc.3", "uuid": "^9.0.1" @@ -57,6 +59,12 @@ "@types/node": "*" } }, + "node_modules/@types/deasync": { + "version": "0.1.5", + "resolved": "https://registry.npmjs.org/@types/deasync/-/deasync-0.1.5.tgz", + "integrity": "sha512-mLov/tw+fOX4ZsrT9xuHOJv8xToOpNsp6W4gp8VDHy2qniJ58izyOzHlisnz5r8HdZ+WItDHtANWZy/W0JEJwg==", + "dev": true + }, "node_modules/@types/express": { "version": "4.17.20", "resolved": "https://registry.npmjs.org/@types/express/-/express-4.17.20.tgz", @@ -213,6 +221,14 @@ } ] }, + "node_modules/bindings": { + "version": "1.5.0", + "resolved": "https://registry.npmjs.org/bindings/-/bindings-1.5.0.tgz", + "integrity": "sha512-p2q/t/mhvuOj/UeLlV6566GD/guowlr0hHxClI0W9m7MWYkL1F0hLo+0Aexs9HSPCtR1SXQ0TD3MMKrXZajbiQ==", + "dependencies": { + "file-uri-to-path": "1.0.0" + } + }, "node_modules/bl": { "version": "5.1.0", "resolved": "https://registry.npmjs.org/bl/-/bl-5.1.0.tgz", @@ -424,6 +440,19 @@ "resolved": "https://registry.npmjs.org/cookie-signature/-/cookie-signature-1.0.6.tgz", "integrity": "sha512-QADzlaHc8icV8I7vbaJXJwod9HWYp8uCqf1xa4OfNu1T7JVxQIrUgOWtHdNDtPiywmFbiS12VjotIXLrKM3orQ==" }, + "node_modules/deasync": { + "version": "0.1.29", + "resolved": "https://registry.npmjs.org/deasync/-/deasync-0.1.29.tgz", + "integrity": "sha512-EBtfUhVX23CE9GR6m+F8WPeImEE4hR/FW9RkK0PMl9V1t283s0elqsTD8EZjaKX28SY1BW2rYfCgNsAYdpamUw==", + "hasInstallScript": true, + "dependencies": { + "bindings": "^1.5.0", + "node-addon-api": "^1.7.1" + }, + "engines": { + "node": ">=0.11.0" + } + }, "node_modules/debug": { "version": "4.3.4", "resolved": "https://registry.npmjs.org/debug/-/debug-4.3.4.tgz", @@ -617,6 +646,11 @@ } ] }, + "node_modules/file-uri-to-path": { + "version": "1.0.0", + "resolved": "https://registry.npmjs.org/file-uri-to-path/-/file-uri-to-path-1.0.0.tgz", + "integrity": "sha512-0Zt+s3L7Vf1biwWZ29aARiVYLx7iMGnEUl9x33fbB/j3jR81u/O2LbqK+Bm1CDSNDKVtJ/YjwY7TUd5SkeLQLw==" + }, "node_modules/finalhandler": { "version": "1.2.0", "resolved": "https://registry.npmjs.org/finalhandler/-/finalhandler-1.2.0.tgz", @@ -1017,6 +1051,11 @@ "node": ">= 0.6" } }, + "node_modules/node-addon-api": { + "version": "1.7.2", + "resolved": "https://registry.npmjs.org/node-addon-api/-/node-addon-api-1.7.2.tgz", + "integrity": "sha512-ibPK3iA+vaY1eEjESkQkM0BbCqFOaZMiXRTtdB0u7b4djtY6JnsjvPdUHVMg6xQt3B8fpTTWHI9A+ADjM9frzg==" + }, "node_modules/number-allocator": { "version": "1.0.14", "resolved": "https://registry.npmjs.org/number-allocator/-/number-allocator-1.0.14.tgz", diff --git a/example/package.json b/example/package.json index 7a41422..d84089c 100644 --- a/example/package.json +++ b/example/package.json @@ -32,6 +32,7 @@ }, "license": "LGPL-3.0-or-later", "devDependencies": { + "@types/deasync": "^0.1.5", "@types/express": "^4.17.20", "typedoc": "^0.25.3", "typedoc-plugin-extras": "^3.0.0", @@ -40,6 +41,7 @@ }, "dependencies": { "aas-multimessagebroker": "file:../lib", + "deasync": "^0.1.29", "express": "^4.18.2", "mqtt": "^5.1.4" } diff --git a/example/src/index.ts b/example/src/index.ts index 367e7f4..709a1b5 100644 --- a/example/src/index.ts +++ b/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"; /** diff --git a/example/src/modules/mqttConnectorVariant.ts b/example/src/modules/mqttConnectorVariant.ts new file mode 100644 index 0000000..eec7d25 --- /dev/null +++ b/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 { + 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 = {}; + + 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): 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 { + 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); + } + } + } + +} \ No newline at end of file