refactor: Only subscribe to needed events not all

No longer subscribe to all events from HA and only subscribe to
state_changed and events define in events: all nodes.

All events can still be subscribed to but doing so may overload the send
queue in HA.

Closes #153

lib/ha-websocket.js

@@ -18,6 +18,9 @@ class HaWebsocket extends EventEmitter {
         this.states = {};
         this.services = {};
         this.statesLoaded = false;
+        this.client = null;
+        this.subscribedEvents = new Set();
+        this.unsubEvents = {};
 
         this.setMaxListeners(0);
     }
@@ -49,6 +52,7 @@ class HaWebsocket extends EventEmitter {
                 self: this,
                 createSocket: this.createSocket
             });
+            this.emit('ha_events:connected');
         } catch (e) {
             this.connectionState = HaWebsocket.DISCONNECTED;
             this.emit('ha_events:close');
@@ -64,8 +68,6 @@ class HaWebsocket extends EventEmitter {
             this.onClientError.bind(this)
         );
 
-        this.client.subscribeEvents(ent => this.onClientEvents(ent));
-
         homeassistant.subscribeEntities(this.client, ent =>
             this.onClientStates(ent)
         );
@@ -76,6 +78,68 @@ class HaWebsocket extends EventEmitter {
         return true;
     }
 
+    async subscribeEvents(events) {
+        const currentEvents = new Set(Object.values(events));
+
+        if (
+            currentEvents.has('__ALL__') &&
+            this.subscribedEvents.has('__ALL__')
+        ) {
+            return;
+        }
+
+        // If events contains '__ALL__' register all events and skip individual ones
+        if (
+            currentEvents.has('__ALL__') &&
+            !this.subscribedEvents.has('__ALL__')
+        ) {
+            this.subscribedEvents.forEach(e => {
+                if (e !== '__ALL__') {
+                    this.unsubEvents[e]();
+                    delete this.unsubEvents[e];
+                }
+            });
+
+            // subscribe to all event and save unsubscribe callback
+            this.unsubEvents.__ALL__ = await this.client.subscribeEvents(ent =>
+                this.onClientEvents(ent)
+            );
+
+            this.subscribedEvents.clear();
+            this.subscribedEvents.add('__ALL__');
+            return;
+        }
+
+        // Always need the state_changed event
+        currentEvents.add('state_changed');
+
+        const add = new Set(
+            [...currentEvents].filter(x => !this.subscribedEvents.has(x))
+        );
+        const remove = new Set(
+            [...this.subscribedEvents].filter(x => !currentEvents.has(x))
+        );
+        const same = new Set(
+            [...currentEvents].filter(x => this.subscribedEvents.has(x))
+        );
+        // Create new subscribed list
+        this.subscribedEvents = new Set([...same, ...add]);
+
+        // Remove unused subscriptions
+        remove.forEach(e => {
+            this.unsubEvents[e]();
+            delete this.unsubEvents[e];
+        });
+
+        // Subscribe to each selected event type and save each unsubscribe callback
+        for (const type of add) {
+            this.unsubEvents[type] = await this.client.subscribeEvents(
+                ent => this.onClientEvents(ent),
+                type
+            );
+        }
+    }
+
     onClientStates(msg) {
         if (!msg || Object.keys(msg).length === 0) {
             return;

lib/node-home-assistant.js

@@ -25,6 +25,7 @@ class HomeAssistant {
         this.config = Object.assign({}, DEFAULTS, config);
         this.http = new HaHttp(this.config);
         this.websocket = new HaWebsocket(this.config);
+        this.registeredEvents = {};
     }
 
     async startListening({ includeRegex, excludeRegex } = {}) {

nodes/config-server/config-server.js

@@ -191,6 +191,10 @@ module.exports = function(RED) {
                     'ha_events:services_loaded',
                     this.onHaServicesLoaded.bind(this)
                 );
+                this.websocket.once(
+                    'ha_events:connected',
+                    this.registerEvents.bind(this)
+                );
             }
         }
 
@@ -269,12 +273,22 @@ module.exports = function(RED) {
                 webSocketClient.close();
             }
         }
+
+        registerEvents() {
+            this.homeAssistant.websocket.subscribeEvents(
+                this.homeAssistant.registeredEvents
+            );
+        }
     }
 
     RED.nodes.registerType('server', ConfigServerNode, {
         credentials: {
-            host: { type: 'text' },
-            access_token: { type: 'text' }
+            host: {
+                type: 'text'
+            },
+            access_token: {
+                type: 'text'
+            }
         }
     });
 };

nodes/events-all/events-all.html

@@ -5,23 +5,32 @@
     category: "home_assistant",
     color: "#038FC7",
     defaults: {
-      name: { value: "" },
-      server: { value: "", type: "server", required: true },
-      event_type: { value: "", required: false }
+      name: {
+        value: ""
+      },
+      server: {
+        value: "",
+        type: "server",
+        required: true
+      },
+      event_type: {
+        value: "",
+        required: false
+      }
     },
     inputs: 0,
     outputs: 1,
     icon: "arrow-right-bold.png",
     paletteLabel: "events: all",
-    label: function() {
+    label: function () {
       return this.name || `events: ${this.event_type || "all"}`;
     },
     labelStyle: nodeVersion.labelStyle,
-    oneditprepare: function() {
+    oneditprepare: function () {
       const NODE = this;
       const $server = $("#node-input-server");
       const utils = {
-        setDefaultServerSelection: function() {
+        setDefaultServerSelection: function () {
           let defaultServer;
           RED.nodes.eachConfig(n => {
             if (n.type === "server" && !defaultServer) defaultServer = n.id;
@@ -33,6 +42,10 @@
       if (!NODE.server) {
         utils.setDefaultServerSelection();
       }
+
+      $("#node-input-event_type").on("change paste keyup", function (e) {
+        $("#eventAlert").toggle(this.value.length === 0)
+      }).trigger();
     }
   });
 </script>
@@ -52,6 +65,8 @@
     <label for="node-input-event_type"><i class="fa fa-tag"></i> Event Type</label>
     <input type="text" id="node-input-event_type" placeholder="leave empty for all events" />
   </div>
+
+  <div id="eventAlert" class="ui-state-error"><p><strong>Alert:</strong> Leaving Event Type empty and listening for all events may overload the websocket message queue.</p></div>
 </script>
 
 <script type="text/x-red" data-help-name="server-events">
@@ -100,4 +115,4 @@ <h3>Client Events</h3>
     <dt>data <span class="property-type optional">string</span></dt>
     <dd>Will contain the error message if event type is type error</dd>
   </dl>
-</script>
+</script>

nodes/events-all/events-all.js

@@ -27,6 +27,11 @@ module.exports = function(RED) {
                     handler: this.onClientServicesLoaded.bind(this)
                 });
             }
+
+            // Registering only needed event types
+            this.nodeConfig.server.homeAssistant.registeredEvents[this.id] =
+                this.nodeConfig.event_type || '__ALL__';
+            this.updateEventList();
         }
 
         onHaEventsAll(evt) {
@@ -64,6 +69,17 @@ module.exports = function(RED) {
             this.clientEvent('services_loaded');
         }
 
+        onClose(nodeRemoved) {
+            super.onClose();
+
+            if (nodeRemoved) {
+                delete this.nodeConfig.server.homeAssistant.registeredEvents[
+                    this.id
+                ];
+                this.updateEventList();
+            }
+        }
+
         onHaEventsClose() {
             super.onHaEventsClose();
             this.clientEvent('disconnected');
@@ -85,6 +101,14 @@ module.exports = function(RED) {
                 this.clientEvent('error', err.message);
             }
         }
+
+        updateEventList() {
+            if (this.isConnected) {
+                this.websocketClient.subscribeEvents(
+                    this.nodeConfig.server.homeAssistant.registeredEvents
+                );
+            }
+        }
     }
 
     RED.nodes.registerType('server-events', ServerEventsNode);

nodes/events-state-changed/events-state-changed.js

@@ -87,6 +87,7 @@ module.exports = function(RED) {
                 if (
                     runAll === undefined &&
                     this.nodeConfig.output_only_on_state_change === true &&
+                    event.old_state &&
                     event.old_state.state === event.new_state.state
                 ) {
                     return null;