Implemented re-subscription if ws connection closes.

Author: dsizominmiro

src/platformAccessory.ts

@@ -5,9 +5,9 @@ import {
   JSONPatchWsIncomingMessage,
   JSONPatchWsOutgoingMessage,
   PrincessHeaterAccessoryContext,
-  PrincessHeaterStateWsIncomingMessage,
+  PrincessHeaterStateWsIncomingMessage, ResponseWsIncomingMessage,
   SubscribeWsOutgoingMessage,
-  WsIncomingMessage,
+  WsIncomingMessage, WsOutgoingMessage,
 } from './ws/types';
 import {WsAPIClient} from './ws';
 import {MessageType} from './ws/const';
@@ -53,12 +53,53 @@ export class HomewizardPrincessHeaterAccessory {
 
     this.platform.log.debug('Subscribing to device updates:', this.accessory.context.device.name);
 
-    wsClient.send<SubscribeWsOutgoingMessage>({
+    this.subscribe().catch(
+      err => this.platform.log.error('Failed to subscribe to device ->', this.accessory.context.device.name, err),
+    );
+  }
+
+  subscribe(): Promise<ResponseWsIncomingMessage> {
+    return this.wsClient.send<SubscribeWsOutgoingMessage>({
       type: MessageType.SubscribeDevice,
       device: this.accessory.context.device.identifier,
     });
   }
 
+  async jsonPatch(path: string, value: boolean | number, callback: CharacteristicSetCallback): Promise<ResponseWsIncomingMessage> {
+
+    const message = {
+      type: MessageType.JSONPatch,
+      device: this.accessory.context.device.identifier,
+      patch: [{
+        op: 'replace',
+        path: path,
+        value,
+      }],
+    };
+
+    const trySend = () => this.wsClient.send(message).then(m => {
+      callback(null);
+      return m;
+    });
+
+    return trySend()
+      .catch(err => {
+        if (err.code === 400) {
+          this.platform.log.warn('Error code 400. Might mean we need to re-subscribe ->', message, err);
+          return this.subscribe().then(() => trySend());
+        } else {
+          this.platform.log.error('Failed to send jsonPatch message ->', message, err);
+          callback(err);
+          throw err;
+        }
+      })
+      .catch(err => {
+        this.platform.log.error('Failed to send jsonPatch message ->', message, err);
+        callback(err);
+        throw err; 
+      });
+  }
+
   onWsMessage(message: WsIncomingMessage) {
     if ('state' in message) {
       this.onStateMessage(message as PrincessHeaterStateWsIncomingMessage);
@@ -146,33 +187,17 @@ export class HomewizardPrincessHeaterAccessory {
 
     this.platform.log.debug('Set Characteristic TargetHeatingCoolingState ->', value);
 
-    this.wsClient.send<JSONPatchWsOutgoingMessage>({
-      type: MessageType.JSONPatch,
-      device: this.accessory.context.device.identifier,
-      patch: [{
-        op: 'replace',
-        path: '/state/power_on',
-        value: value === this.platform.Characteristic.TargetHeatingCoolingState.HEAT,
-      }],
-    })
-      .then(() => callback(null))
-      .catch((err) => callback(err));
+    this.jsonPatch(
+      '/state/power_on',
+      value === this.platform.Characteristic.TargetHeatingCoolingState.HEAT,
+      callback,
+    );
   }
 
   setTargetTemperature(value: CharacteristicValue, callback: CharacteristicSetCallback) {
 
     this.platform.log.debug('Set Characteristic TargetTemperature ->', value);
 
-    this.wsClient.send<JSONPatchWsOutgoingMessage>({
-      type: MessageType.JSONPatch,
-      device: this.accessory.context.device.identifier,
-      patch: [{
-        op: 'replace',
-        path: '/state/target_temperature',
-        value: value as number,
-      }],
-    })
-      .then(() => callback(null))
-      .catch((err) => callback(err));
+    this.jsonPatch('/state/target_temperature', value as number, callback);
   }
 }

src/ws/index.ts

@@ -1,5 +1,5 @@
 import WebSocket from 'ws';
-import {HelloWsOutgoingMessage, WsIncomingMessage, WsOutgoingMessage} from './types';
+import {HelloWsOutgoingMessage, ResponseWsIncomingMessage, WsIncomingMessage, WsOutgoingMessage} from './types';
 import {MessageType, WS_URL} from './const';
 import {EventEmitter} from 'events';
 import {Logger} from 'homebridge';
@@ -21,7 +21,7 @@ export class WsAPIClient extends EventEmitter {
 
   public async send<M extends WsOutgoingMessage>(
     message: Omit<M, 'message_id'>,
-  ): Promise<M> {
+  ): Promise<ResponseWsIncomingMessage> {
 
     const ws = await this._getWebSocket();
     return this._send<M>(message, ws);
@@ -99,7 +99,7 @@ export class WsAPIClient extends EventEmitter {
   private async _send<M extends WsOutgoingMessage>(
     message: Omit<M, 'message_id'>,
     ws: WebSocket,
-  ): Promise<M> {
+  ): Promise<ResponseWsIncomingMessage> {
 
     const sentMessage = await new Promise((res, rej) => {
       const messageId = ++this.lastMessageId;
@@ -127,7 +127,7 @@ export class WsAPIClient extends EventEmitter {
 
   private _waitForResponse<M extends WsOutgoingMessage>(
     outgoingMessage: M,
-  ): Promise<M> {
+  ): Promise<ResponseWsIncomingMessage> {
     return new Promise((res, rej) => {
 
       this.log.debug('Waiting for message response -> ', outgoingMessage);
@@ -145,7 +145,7 @@ export class WsAPIClient extends EventEmitter {
           this.off('message', onMessage);
           clearTimeout(timeout);
           if (incomingMessage.status === 200) {
-            res(outgoingMessage);
+            res(incomingMessage as ResponseWsIncomingMessage);
           } else {
             rej(incomingMessage);
           }