Throttle event emits (#55)

Digital-Alchemy-TS · Jul 15, 2024 · 2f282d4 · 2f282d4
1 parent 9b8b6d7
commit 2f282d4
Show file tree

Hide file tree

Showing 19 changed files with 255 additions and 26 deletions.
diff --git a/package.json b/package.json
@@ -2,7 +2,7 @@
   "name": "@digital-alchemy/hass",
   "repository": "https://github.com/Digital-Alchemy-TS/hass",
   "homepage": "https://docs.digital-alchemy.app",
-  "version": "24.7.4",
+  "version": "24.7.5",
   "scripts": {
     "build": "rm -rf dist/; tsc",
     "lint": "eslint src",

diff --git a/src/e2e/area.spec.ts b/src/e2e/area.spec.ts
@@ -7,7 +7,7 @@ import {
 } from "@digital-alchemy/core";
 
 import { TAreaId } from "../dynamic";
-import { AREA_REGISTRY_UPDATED } from "../helpers";
+import { AREA_REGISTRY_UPDATED, AreaDetails } from "../helpers";
 import { CreateTestingApplication, SILENT_BOOT } from "../mock_assistant";
 import { BASE_URL, TOKEN } from "./utils";
 
@@ -34,7 +34,7 @@ describe("Area E2E", () => {
           let hit = false;
           event.on(AREA_REGISTRY_UPDATED, () => (hit = true));
           await hass.socket.fireEvent("area_registry_updated");
-          await sleep(50);
+          await sleep(100);
           expect(hit).toBe(true);
           await application.teardown();
         });
@@ -68,7 +68,7 @@ describe("Area E2E", () => {
     application = CreateTestingApplication({
       Test({ lifecycle, hass }: TServiceParams) {
         lifecycle.onReady(async () => {
-          const item = hass.area.current.find(i => i.area_id === testArea);
+          const item = hass.area.current.find(i => i.area_id === testArea) as AreaDetails;
           await hass.area.update({
             ...item,
             name: "extra test",

diff --git a/src/e2e/entity.spec.ts b/src/e2e/entity.spec.ts
@@ -78,7 +78,7 @@ describe("Entity E2E", () => {
               await hass.call.switch.toggle({
                 entity_id: "switch.porch_light",
               });
-              await sleep(50);
+              await sleep(100);
             }
             expect(counter).toBe(expected);
             await application.teardown();
@@ -103,7 +103,7 @@ describe("Entity E2E", () => {
               await hass.call.switch.toggle({
                 entity_id: "switch.porch_light",
               });
-              await sleep(50);
+              await sleep(100);
             }
             expect(counter).toBe(1);
             await application.teardown();
@@ -128,7 +128,7 @@ describe("Entity E2E", () => {
               await hass.call.switch.toggle({
                 entity_id: "switch.porch_light",
               });
-              await sleep(50);
+              await sleep(100);
             }
             expect(counter).toBe(1);
             await application.teardown();
@@ -153,7 +153,7 @@ describe("Entity E2E", () => {
               await hass.call.switch.toggle({
                 entity_id: "switch.porch_light",
               });
-              await sleep(50);
+              await sleep(100);
             }
             expect(counter).toBe(1);
             await application.teardown();
@@ -175,7 +175,7 @@ describe("Entity E2E", () => {
               await hass.call.switch.toggle({
                 entity_id: "switch.porch_light",
               });
-              await sleep(50);
+              await sleep(100);
             }
             expect(counter).toBe(1);
             await application.teardown();
@@ -196,7 +196,7 @@ describe("Entity E2E", () => {
               expect(next.state).toBe(porch.state);
             });
             await hass.call.switch.toggle({ entity_id: "switch.porch_light" });
-            await sleep(50);
+            await sleep(100);
             await application.teardown();
           });
         },
@@ -215,7 +215,7 @@ describe("Entity E2E", () => {
               expect(next.state).toBe(porch.state);
             });
             await hass.call.switch.toggle({ entity_id: "switch.porch_light" });
-            await sleep(50);
+            await sleep(100);
             await application.teardown();
           });
         },

diff --git a/src/e2e/floor.spec.ts b/src/e2e/floor.spec.ts
@@ -34,7 +34,7 @@ describe("Floor E2E", () => {
           let hit = false;
           event.on(FLOOR_REGISTRY_UPDATED, () => (hit = true));
           await hass.socket.fireEvent("floor_registry_updated");
-          await sleep(50);
+          await sleep(100);
           expect(hit).toBe(true);
           await application.teardown();
         });

diff --git a/src/e2e/label.spec.ts b/src/e2e/label.spec.ts
@@ -34,7 +34,7 @@ describe("Label E2E", () => {
           let hit = false;
           event.on(LABEL_REGISTRY_UPDATED, () => (hit = true));
           await hass.socket.fireEvent("label_registry_updated");
-          await sleep(50);
+          await sleep(100);
           expect(hit).toBe(true);
           await application.teardown();
         });

diff --git a/src/extensions/area.extension.ts b/src/extensions/area.extension.ts
@@ -1,6 +1,7 @@
 import {
   eachSeries,
   InternalError,
+  throttle,
   TServiceParams,
 } from "@digital-alchemy/core";
 
@@ -38,6 +39,7 @@ export function Area({
     context,
     event_type: "area_registry_updated",
     async exec() {
+      await throttle(AREA_REGISTRY_UPDATED, config.hass.EVENT_THROTTLE_MS);
       hass.area.current = await hass.area.list();
       logger.debug(`area registry updated`);
       event.emit(AREA_REGISTRY_UPDATED);

diff --git a/src/extensions/device.extension.ts b/src/extensions/device.extension.ts
@@ -1,4 +1,4 @@
-import { TServiceParams } from "@digital-alchemy/core";
+import { throttle, TServiceParams } from "@digital-alchemy/core";
 
 import {
   DEVICE_REGISTRY_UPDATED,
@@ -31,6 +31,7 @@ export function Device({
     context,
     event_type: "device_registry_updated",
     async exec() {
+      await throttle(DEVICE_REGISTRY_UPDATED, config.hass.EVENT_THROTTLE_MS);
       hass.device.current = await hass.device.list();
       logger.debug(`device registry updated`);
       event.emit(DEVICE_REGISTRY_UPDATED);

diff --git a/src/extensions/entity.extension.ts b/src/extensions/entity.extension.ts
@@ -6,6 +6,7 @@ import {
   SECOND,
   sleep,
   START,
+  throttle,
   TServiceParams,
 } from "@digital-alchemy/core";
 import dayjs, { Dayjs } from "dayjs";
@@ -296,22 +297,23 @@ export function EntityManager({
       type: "config/entity_registry/get",
     });
   }
+  hass.socket.subscribe({
+    context,
+    event_type: "entity_registry_updated",
+    async exec() {
+      await throttle(ENTITY_REGISTRY_UPDATED, config.hass.EVENT_THROTTLE_MS);
+      logger.debug("entity registry updated");
+      hass.entity.registry.current = await hass.entity.registry.list();
+      event.emit(ENTITY_REGISTRY_UPDATED);
+    },
+  });
 
   // #MARK: onConnect
   hass.socket.onConnect(async () => {
     if (!config.hass.AUTO_CONNECT_SOCKET || !config.hass.MANAGE_REGISTRY) {
       return;
     }
     hass.entity.registry.current = await hass.entity.registry.list();
-    hass.socket.subscribe({
-      context,
-      event_type: "entity_registry_updated",
-      async exec() {
-        logger.debug("entity registry updated");
-        hass.entity.registry.current = await hass.entity.registry.list();
-        event.emit(ENTITY_REGISTRY_UPDATED);
-      },
-    });
   });
 
   // #MARK: RemoveEntity

diff --git a/src/extensions/floor.extension.ts b/src/extensions/floor.extension.ts
@@ -1,4 +1,4 @@
-import { TServiceParams } from "@digital-alchemy/core";
+import { throttle, TServiceParams } from "@digital-alchemy/core";
 
 import { TFloorId } from "../dynamic";
 import {
@@ -32,6 +32,7 @@ export function Floor({
     context,
     event_type: "floor_registry_updated",
     async exec() {
+      await throttle(FLOOR_REGISTRY_UPDATED, config.hass.EVENT_THROTTLE_MS);
       hass.floor.current = await hass.floor.list();
       logger.debug(`floor registry updated`);
       event.emit(FLOOR_REGISTRY_UPDATED);

diff --git a/src/extensions/label.extension.ts b/src/extensions/label.extension.ts
@@ -1,4 +1,4 @@
-import { TServiceParams } from "@digital-alchemy/core";
+import { throttle, TServiceParams } from "@digital-alchemy/core";
 
 import { TLabelId } from "../dynamic";
 import {
@@ -32,6 +32,7 @@ export function Label({
     context,
     event_type: "label_registry_updated",
     async exec() {
+      await throttle(LABEL_REGISTRY_UPDATED, config.hass.EVENT_THROTTLE_MS);
       hass.label.current = await hass.label.list();
       logger.debug(`label registry updated`);
       event.emit(LABEL_REGISTRY_UPDATED);

diff --git a/src/extensions/websocket-api.extension.ts b/src/extensions/websocket-api.extension.ts
@@ -675,6 +675,11 @@ export function WebsocketAPI({
      */
     setConnectionState,
 
+    /**
+     * internal
+     */
+    socketEvents,
+
     /**
      * Subscribe to hass core registry updates.
      *

diff --git a/src/extensions/zone.extension.ts b/src/extensions/zone.extension.ts
@@ -1,4 +1,4 @@
-import { TServiceParams } from "@digital-alchemy/core";
+import { throttle, TServiceParams } from "@digital-alchemy/core";
 
 import {
   EARLY_ON_READY,
@@ -32,6 +32,7 @@ export function Zone({
     context,
     event_type: "zone_registry_updated",
     async exec() {
+      await throttle(ZONE_REGISTRY_UPDATED, config.hass.EVENT_THROTTLE_MS);
       hass.zone.current = await hass.zone.list();
       logger.debug(`zone registry updated`);
       event.emit(ZONE_REGISTRY_UPDATED);

diff --git a/src/hass.module.ts b/src/hass.module.ts
@@ -47,6 +47,11 @@ export const LIB_HASS = CreateLibrary({
       enum: ["prefer", "forbid", "allow"],
       type: "string",
     } as StringConfig<AllowRestOptions>,
+    EVENT_THROTTLE_MS: {
+      default: 50,
+      description: "Throttle reactions to registry changes",
+      type: "number",
+    },
     EXPECT_RESPONSE_AFTER: {
       default: 5,
       description:

diff --git a/src/testing/area.spec.ts b/src/testing/area.spec.ts
@@ -2,6 +2,7 @@ import {
   ApplicationDefinition,
   OptionalModuleConfiguration,
   ServiceMap,
+  sleep,
   TServiceParams,
 } from "@digital-alchemy/core";
 
@@ -113,6 +114,40 @@ describe("Area", () => {
         );
       });
 
+      it("should throttle updates properly", async () => {
+        expect.assertions(1);
+        application = CreateTestingApplication({
+          Test({ lifecycle, hass }: TServiceParams) {
+            jest
+              .spyOn(hass.socket, "sendMessage")
+              .mockImplementation(async () => undefined);
+            let counter = 0;
+            hass.events.onAreaRegistryUpdate(() => counter++);
+            lifecycle.onReady(async () => {
+              setImmediate(async () => {
+                hass.socket.socketEvents.emit("area_registry_updated");
+                await sleep(5);
+                hass.socket.socketEvents.emit("area_registry_updated");
+                await sleep(5);
+                hass.socket.socketEvents.emit("area_registry_updated");
+                await sleep(75);
+                hass.socket.socketEvents.emit("area_registry_updated");
+              });
+              await sleep(200);
+              expect(counter).toBe(2);
+            });
+          },
+        });
+        await application.bootstrap(
+          SILENT_BOOT({
+            hass: {
+              AUTO_CONNECT_SOCKET: false,
+              AUTO_SCAN_CALL_PROXY: false,
+            },
+          }),
+        );
+      });
+
       it("should call delete properly", async () => {
         expect.assertions(1);
         application = CreateTestingApplication({

diff --git a/src/testing/device.spec.ts b/src/testing/device.spec.ts
@@ -2,6 +2,7 @@ import {
   ApplicationDefinition,
   OptionalModuleConfiguration,
   ServiceMap,
+  sleep,
   TServiceParams,
 } from "@digital-alchemy/core";
 
@@ -71,6 +72,40 @@ describe("Device", () => {
     });
   });
 
+  it("should throttle updates properly", async () => {
+    expect.assertions(1);
+    application = CreateTestingApplication({
+      Test({ lifecycle, hass }: TServiceParams) {
+        jest
+          .spyOn(hass.socket, "sendMessage")
+          .mockImplementation(async () => undefined);
+        let counter = 0;
+        hass.events.onDeviceRegistryUpdate(() => counter++);
+        lifecycle.onReady(async () => {
+          setImmediate(async () => {
+            hass.socket.socketEvents.emit("device_registry_updated");
+            await sleep(5);
+            hass.socket.socketEvents.emit("device_registry_updated");
+            await sleep(5);
+            hass.socket.socketEvents.emit("device_registry_updated");
+            await sleep(75);
+            hass.socket.socketEvents.emit("device_registry_updated");
+          });
+          await sleep(200);
+          expect(counter).toBe(2);
+        });
+      },
+    });
+    await application.bootstrap(
+      SILENT_BOOT({
+        hass: {
+          AUTO_CONNECT_SOCKET: false,
+          AUTO_SCAN_CALL_PROXY: false,
+        },
+      }),
+    );
+  });
+
   describe("API", () => {
     describe("Formatting", () => {
       it("should call list properly", async () => {