mqtt: support invoking methods

plugins/mqtt/package-lock.json

@@ -1,12 +1,12 @@
 {
    "name": "@scrypted/mqtt",
-   "version": "0.0.68",
+   "version": "0.0.69",
    "lockfileVersion": 2,
    "requires": true,
    "packages": {
       "": {
          "name": "@scrypted/mqtt",
-         "version": "0.0.68",
+         "version": "0.0.69",
          "dependencies": {
             "@types/node": "^16.6.1",
             "aedes": "^0.46.1",

plugins/mqtt/package.json

@@ -41,5 +41,5 @@
       "@scrypted/common": "file:../../common",
       "@types/nunjucks": "^3.2.0"
    },
-   "version": "0.0.68"
+   "version": "0.0.69"
 }

plugins/mqtt/src/main.ts

@@ -260,8 +260,20 @@ class MqttPublisherMixin extends SettingsMixinDeviceBase<any> {
         });
         client.setMaxListeners(Infinity);
 
+        const allProperties: string[] = [];
+        const allMethods: string[] = [];
+        for (const iface of this.device.interfaces) {
+            const methods = ScryptedInterfaceDescriptors[iface]?.methods || [];
+            allMethods.push(...methods);
+            const properties = ScryptedInterfaceDescriptors[iface]?.properties || [];
+            allProperties.push(...properties);
+        }
+
         client.on('connect', packet => {
             this.console.log('MQTT client connected, publishing current state.');
+            for (const method of allMethods) {
+                client.subscribe(this.pathname + '/' + method);
+            }
 
             for (const iface of this.device.interfaces) {
                 for (const prop of ScryptedInterfaceDescriptors[iface]?.properties || []) {
@@ -278,6 +290,22 @@ class MqttPublisherMixin extends SettingsMixinDeviceBase<any> {
             this.console.log('mqtt client error', e);
         });
 
+        client.on('message', async (messageTopic, message) => {
+            const method = messageTopic.substring(this.pathname.length + 1);
+            if (!allMethods.includes(method)) {
+                if (!allProperties.includes(method))
+                    this.console.warn('unknown topic', method);
+                return;
+            }
+            try {
+                const args = JSON.parse(message.toString());
+                await this.device[method](...args);
+            }
+            catch (e) {
+                this.console.warn('error invoking method', e);
+            }
+        });
+
         return this.client;
     }
 
@@ -482,7 +510,7 @@ class MqttProvider extends ScryptedDeviceBase implements DeviceProvider, Setting
     }
 
     async releaseDevice(id: string, nativeId: string): Promise<void> {
-        
+
     }
 
     createMqttDevice(nativeId: string): MqttDevice {