add command queue with delays

README.md

@@ -85,7 +85,8 @@ Basic example of the configuration (config.json) file.
   "server": {
     "port": 3000,
     "deviceReconnectWait": 5000,
-    "refreshTime": 60000
+    "refreshTime": 60000,
+    "sendCommandDelay": 5000
   },
   "webhooks": ["http://localhost:4321/webhook"],
   "devices": [

config.example.json

@@ -2,7 +2,8 @@
   "server": {
     "port": 3000,
     "deviceReconnectWait": 5000,
-    "refreshTime": 60000
+    "refreshTime": 60000,
+    "sendCommandDelay": 5000
   },
   "webhooks": ["http://localhost:4321/webhook"],
   "devices": [

package.json

@@ -11,11 +11,13 @@
   "author": "Christopher Jansson",
   "license": "MIT",
   "dependencies": {
+    "async-lock": "^1.4.0",
     "express": "^4.18.2",
     "node-fetch": "^3.3.0",
     "tuyapi": "github:codetheweb/tuyapi"
   },
   "devDependencies": {
+    "@types/async-lock": "^1.4.2",
     "@types/express": "^4.17.17",
     "@vercel/ncc": "^0.36.1",
     "nodemon": "^2.0.20",

src/action-queue.ts

@@ -0,0 +1,59 @@
+import AsyncLock from "async-lock";
+import ActiveDevice from "./active-device";
+import Configuration from "./configuration";
+import {
+  ITuyaMultipleProperties,
+  ITuyaSingleProperty,
+} from "./interfaces/tuya";
+
+let processQueueTimeout = undefined;
+let queue: QueueItem[] = [];
+
+let lock = new AsyncLock();
+
+const queueAction = (
+  device: ActiveDevice,
+  data: ITuyaMultipleProperties | ITuyaSingleProperty
+) => {
+  lock.acquire("processQueue", () => {
+    queue.push({ device, data });
+    startTimeout();
+  });
+};
+
+const startTimeout = () => {
+  lock.acquire("processQueueTimeout", () => {
+    if (!processQueueTimeout) {
+      processQueueTimeout = setTimeout(
+        processQueueItem,
+        Configuration.instance.server().sendCommandDelay
+      );
+    }
+  });
+};
+
+const processQueueItem = async () => {
+  let queueItem: QueueItem;
+  await lock.acquire("processQueue", () => {
+    queueItem = queue.pop();
+  });
+
+  if (queueItem) {
+    await queueItem.device.set(queueItem.data);
+  }
+
+  await lock.acquire("processQueueTimeout", () => {
+    processQueueTimeout = undefined;
+  });
+
+  if (queueItem) {
+    startTimeout();
+  }
+};
+
+interface QueueItem {
+  device: ActiveDevice;
+  data: ITuyaMultipleProperties | ITuyaSingleProperty;
+}
+
+export default queueAction;

src/active-device.ts

@@ -79,6 +79,7 @@ export default class ActiveDevice {
       return;
     }
 
+    Logger.Debug(`${this.tuya.device.id} - sending command`, data);
     let response = await this.tuya.set(data);
     return response ? response.dps : {};
   }

src/app.ts

@@ -9,6 +9,7 @@ import Logger from "./logger";
 import processDevices from "./process-devices";
 import PublicDevice from "./public-device.model";
 import refreshDevices from "./refresh-devices";
+import queueAction from "./action-queue";
 
 Logger.Info(`Starting`);
 
@@ -87,8 +88,8 @@ app.post("/devices/:deviceId/send", async (req, res) => {
       }
     }
 
-    let result = await device.set(<ITuyaSingleProperty>action);
-    res.json(result);
+    queueAction(device, <ITuyaSingleProperty>action);
+    res.sendStatus(200);
   } else if (action.hasOwnProperty("data")) {
     for (var key in action) {
       if (action.hasOwnProperty(key) && key !== "data") {
@@ -97,8 +98,8 @@ app.post("/devices/:deviceId/send", async (req, res) => {
     }
 
     action.multiple = true;
-    let result = await device.set(<ITuyaMultipleProperties>action);
-    res.json(result);
+    queueAction(device, <ITuyaMultipleProperties>action);
+    res.sendStatus(200);
   } else {
     res.sendStatus(400);
   }

src/configuration.ts

@@ -9,6 +9,7 @@ export default class Configuration {
     server: {
       port: 3000,
       deviceReconnectWait: 5000,
+      sendCommandDelay: 5000,
     },
     webhooks: [],
     devices: [],

src/interfaces/config/server.ts

@@ -2,4 +2,5 @@ export default interface IConfigServer {
   port: number;
   deviceReconnectWait: number;
   refreshTime?: number;
+  sendCommandDelay: number;
 }