websocket reconnect

.gitignore

@@ -5,6 +5,8 @@ tests/
 
 digitalstrom-mqtt
 
+node_modules/
+
 docs/docker/homeassistant
 .env
 /config.yaml

docs/websocket-server/index.js

@@ -0,0 +1,18 @@
+import { WebSocketServer } from 'ws';
+
+/**
+ * This websocket server is here just to simulate a digitalstrom server and how to reconnect
+ * @type {WebSocketServer}
+ */
+
+const wss = new WebSocketServer({ port: 8090 });
+
+wss.on('connection', function connection(ws) {
+    ws.on('message', function message(data) {
+        console.log('received: %s', data);
+    });
+
+    // ws.send('something');
+});
+
+console.log('Started')

docs/websocket-server/package.json

@@ -0,0 +1,6 @@
+{
+  "type": "module",
+  "dependencies": {
+    "ws": "^8.16.0"
+  }
+}

pkg/digitalstrom/client.go

@@ -13,8 +13,11 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"time"
 )
 
+const WEBSOCKET_RECONNECT_DELAY = 5 * time.Second
+
 type NotificationCallback func(notification WebsocketNotification)
 
 // Client is the interface definition as used by this library, the
@@ -43,9 +46,10 @@ type Client interface {
 // client implements the DigitalStrom interface.
 // Clients are safe for concurrent use by multiple goroutines.
 type client struct {
-	httpClient          *http.Client
-	options             ClientOptions
-	websocketConnection *websocket.Conn
+	httpClient              *http.Client
+	options                 ClientOptions
+	websocketConnection     *websocket.Conn
+	websocketConnectionOpen bool
 
 	notificationCallbacks map[string]NotificationCallback
 }
@@ -67,8 +71,7 @@ func NewClient(options *ClientOptions) Client {
 	}
 }
 
-func (c *client) Connect() error {
-	// TODO handle reconnection
+func (c *client) websocketConnect() error {
 	websocketHost := "ws://" + c.options.Host + ":8090/api/v1/apartment/notifications"
 	log.Trace().Str("host", websocketHost).Msg("Connecting to websocket")
 	headers := http.Header{}
@@ -86,15 +89,36 @@ func (c *client) Connect() error {
 	if err != nil {
 		return fmt.Errorf("error writing to websocket: %w", err)
 	}
+	log.Info().Msg("Connected to websocket for notifications")
+	c.websocketConnectionOpen = true
+	return nil
+}
 
+func (c *client) Connect() error {
+	c.websocketConnectionOpen = false
+	err := c.websocketConnect()
+	if err != nil {
+		return err
+	}
+
+	// websocket message reader
 	go func() {
 		firstMessage := true
 		for {
 			var notification WebsocketNotification
 			err := c.websocketConnection.ReadJSON(&notification)
 			if err != nil {
-				log.Error().Err(err).Msg("Websocket reading error")
-				break
+				if !c.websocketConnectionOpen {
+					// we're closing, ignore read errors
+					break
+				} else {
+					log.Error().Err(err).Msg("Websocket reading error, will try to reconnect")
+					time.Sleep(WEBSOCKET_RECONNECT_DELAY)
+					err = c.websocketConnect()
+					if err != nil {
+						log.Error().Err(err).Msg("Websocket reconnect error")
+					}
+				}
 			} else if notification.Arguments == nil || len(notification.Arguments) == 0 {
 				if !firstMessage {
 					log.Warn().Msg("No argument received in notification")
@@ -107,18 +131,17 @@ func (c *client) Connect() error {
 			}
 			firstMessage = false
 		}
-		log.Warn().Msg("Closing websocket reader")
+		log.Info().Msg("Closing websocket reader")
 	}()
 
 	return nil
 }
 
 // Disconnect stops all the ongoing calls and unsubscribe from the notification websocket
 func (c *client) Disconnect() error {
-
+	c.websocketConnectionOpen = false
 	c.httpClient.CloseIdleConnections()
-	c.websocketConnection.Close()
-
+	_ = c.websocketConnection.Close()
 	return nil
 }