[neohub] Improved Web-Socket Communications (#16312)

* [neohub] Improved WebSocket Communications
* [neohub] session recycled only by handler; not by socket class

---------

Signed-off-by: Andrew Fiddian-Green <software@whitebear.ch>
Signed-off-by: Ciprian Pascu <contact@ciprianpascu.ro>

bundles/org.openhab.binding.neohub/src/main/java/org/openhab/binding/neohub/internal/NeoHubSocket.java

@@ -47,8 +47,7 @@ public class NeoHubSocket extends NeoHubSocketBase {
         IOException caughtException = null;
         StringBuilder builder = new StringBuilder();
 
-        throttle();
+        try (Socket socket = new Socket(); Throttler throttler = new Throttler();) {
-        try (Socket socket = new Socket()) {
             int port = config.portNumber > 0 ? config.portNumber : NeoHubBindingConstants.PORT_TCP;
             socket.connect(new InetSocketAddress(config.hostName, port), config.socketTimeout * 1000);
             socket.setSoTimeout(config.socketTimeout * 1000);
@@ -76,6 +75,8 @@ public class NeoHubSocket extends NeoHubSocketBase {
         } catch (IOException e) {
             // catch IOExceptions here, and save them to be re-thrown later
             caughtException = e;
+        } catch (InterruptedException e) {
+            caughtException = new IOException(e);
         }
 
         String responseJson = builder.toString().strip();

bundles/org.openhab.binding.neohub/src/main/java/org/openhab/binding/neohub/internal/NeoHubSocketBase.java

@@ -17,6 +17,8 @@ import java.io.IOException;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.Optional;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 import org.eclipse.jdt.annotation.NonNullByDefault;
 
@@ -33,6 +35,7 @@ public abstract class NeoHubSocketBase implements Closeable {
     protected final String hubId;
 
     private static final int REQUEST_INTERVAL_MILLISECS = 1000;
+    private final Lock lock = new ReentrantLock(true);
     private Optional<Instant> lastRequestTime = Optional.empty();
 
     public NeoHubSocketBase(NeoHubConfiguration config, String hubId) {
@@ -51,22 +54,31 @@ public abstract class NeoHubSocketBase implements Closeable {
     public abstract String sendMessage(final String requestJson) throws IOException, NeoHubException;
 
     /**
-     * Method for throttling requests to prevent overloading the hub.
+     * Class for throttling requests to prevent overloading the hub.
      * <p>
      * The NeoHub can get confused if, while it is uploading data to the cloud, it also receives too many local
      * requests, so this method throttles the requests to one per REQUEST_INTERVAL_MILLISECS maximum.
      *
-     * @throws NeoHubException if the wait is interrupted
+     * @throws InterruptedException if the wait is interrupted
      */
-    protected synchronized void throttle() throws NeoHubException {
+    protected class Throttler implements AutoCloseable {
-        try {
+
-            Instant now = Instant.now();
+        public Throttler() throws InterruptedException {
-            long delay = lastRequestTime
+            lock.lock();
-                    .map(t -> Math.max(0, Duration.between(now, t).toMillis() + REQUEST_INTERVAL_MILLISECS)).orElse(0L);
+            long delay;
-            lastRequestTime = Optional.of(now.plusMillis(delay));
+            synchronized (NeoHubSocketBase.this) {
+                Instant now = Instant.now();
+                delay = lastRequestTime
+                        .map(t -> Math.max(0, Duration.between(now, t).toMillis() + REQUEST_INTERVAL_MILLISECS))
+                        .orElse(0L);
+                lastRequestTime = Optional.of(now.plusMillis(delay));
+            }
             Thread.sleep(delay);
-        } catch (InterruptedException e) {
+        }
-            throw new NeoHubException("Throttle sleep interrupted", e);
+
+        @Override
+        public void close() {
+            lock.unlock();
         }
     }
 }

bundles/org.openhab.binding.neohub/src/main/java/org/openhab/binding/neohub/internal/NeoHubTlsTrustManagerProvider.java

@@ -0,0 +1,45 @@
+/**
+ * Copyright (c) 2010-2024 Contributors to the openHAB project
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.openhab.binding.neohub.internal;
+
+import javax.net.ssl.X509ExtendedTrustManager;
+
+import org.eclipse.jdt.annotation.NonNullByDefault;
+import org.openhab.core.io.net.http.TlsTrustManagerProvider;
+import org.openhab.core.io.net.http.TrustAllTrustManager;
+
+/**
+ * A {@link TlsTrustManagerProvider} implementation to validate the NeoHub web socket self signed certificate.
+ *
+ * @author Andrew Fiddian-Green - Initial contribution
+ */
+@NonNullByDefault
+public class NeoHubTlsTrustManagerProvider implements TlsTrustManagerProvider {
+
+    private final String fullHostName;
+
+    public NeoHubTlsTrustManagerProvider(NeoHubConfiguration config) {
+        fullHostName = String.format("%s:%d", config.hostName,
+                config.portNumber > 0 ? config.portNumber : NeoHubBindingConstants.PORT_WSS);
+    }
+
+    @Override
+    public String getHostName() {
+        return fullHostName;
+    }
+
+    @Override
+    public X509ExtendedTrustManager getTrustManager() {
+        return TrustAllTrustManager.getInstance();
+    }
+}

bundles/org.openhab.binding.neohub/src/main/java/org/openhab/binding/neohub/internal/NeoHubWebSocket.java

@@ -16,11 +16,12 @@ import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.time.Instant;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ExecutionException;
 
 import org.eclipse.jdt.annotation.NonNullByDefault;
 import org.eclipse.jdt.annotation.Nullable;
-import org.eclipse.jetty.util.ssl.SslContextFactory;
 import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose;
 import org.eclipse.jetty.websocket.api.annotations.OnWebSocketConnect;
@@ -28,9 +29,12 @@ import org.eclipse.jetty.websocket.api.annotations.OnWebSocketError;
 import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage;
 import org.eclipse.jetty.websocket.api.annotations.WebSocket;
 import org.eclipse.jetty.websocket.client.WebSocketClient;
+import org.openhab.core.io.net.http.TlsTrustManagerProvider;
 import org.openhab.core.io.net.http.WebSocketFactory;
 import org.openhab.core.thing.ThingUID;
 import org.openhab.core.thing.util.ThingWebClientUtil;
+import org.osgi.framework.FrameworkUtil;
+import org.osgi.framework.ServiceRegistration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -55,10 +59,12 @@ public class NeoHubWebSocket extends NeoHubSocketBase {
     private final Logger logger = LoggerFactory.getLogger(NeoHubWebSocket.class);
     private final Gson gson = new Gson();
     private final WebSocketClient webSocketClient;
+    private final ServiceRegistration<?> trustManagerRegistration;
 
     private @Nullable Session session = null;
-    private String responseOuter = "";
+    private @Nullable IOException websocketException = null;
-    private boolean responsePending;
+    private List<String> responses = new CopyOnWriteArrayList<>();
+    private boolean closing;
 
     /**
      * DTO to receive and parse the response JSON.
@@ -78,16 +84,21 @@ public class NeoHubWebSocket extends NeoHubSocketBase {
             throws IOException {
         super(config, bridgeUID.getAsString());
 
-        SslContextFactory.Client sslContextFactory = new SslContextFactory.Client();
-        sslContextFactory.setTrustAll(true);
         String name = ThingWebClientUtil.buildWebClientConsumerName(bridgeUID, null);
-        webSocketClient = webSocketFactory.createWebSocketClient(name, sslContextFactory);
+        webSocketClient = webSocketFactory.createWebSocketClient(name);
         webSocketClient.setConnectTimeout(config.socketTimeout * 1000);
         try {
             webSocketClient.start();
         } catch (Exception e) {
             throw new IOException("Error starting Web Socket client", e);
         }
+        NeoHubTlsTrustManagerProvider trustManagerProvider = new NeoHubTlsTrustManagerProvider(config);
+        try {
+            trustManagerRegistration = FrameworkUtil.getBundle(getClass()).getBundleContext()
+                    .registerService(TlsTrustManagerProvider.class.getName(), trustManagerProvider, null);
+        } catch (IllegalStateException e) {
+            throw new IOException("Error registering trust manager", e);
+        }
     }
 
     /**
@@ -95,14 +106,13 @@ public class NeoHubWebSocket extends NeoHubSocketBase {
      *
      * @throws IOException if unable to open the web socket
      */
-    private void startSession() throws IOException {
+    private synchronized void startSession() throws IOException {
         Session session = this.session;
         if (session == null || !session.isOpen()) {
-            closeSession();
             try {
                 int port = config.portNumber > 0 ? config.portNumber : NeoHubBindingConstants.PORT_WSS;
                 URI uri = new URI(String.format("wss://%s:%d", config.hostName, port));
-                webSocketClient.connect(this, uri).get();
+                this.session = webSocketClient.connect(this, uri).get();
             } catch (InterruptedException e) {
                 Thread.currentThread().interrupt();
                 throw new IOException("Error starting session", e);
@@ -112,17 +122,6 @@ public class NeoHubWebSocket extends NeoHubSocketBase {
         }
     }
 
-    /**
-     * Close the web socket session.
-     */
-    private void closeSession() {
-        Session session = this.session;
-        this.session = null;
-        if (session != null) {
-            session.close();
-        }
-    }
-
     /**
      * Helper to escape the quote marks in a JSON string.
      *
@@ -154,85 +153,101 @@ public class NeoHubWebSocket extends NeoHubSocketBase {
     }
 
     @Override
-    public synchronized String sendMessage(final String requestJson) throws IOException, NeoHubException {
+    public String sendMessage(final String requestJson) throws IOException, NeoHubException {
-        // start the session
+        if (!closing && websocketException != null) {
-        startSession();
+            throw websocketException;
-
-        // session start failed
-        Session session = this.session;
-        if (session == null) {
-            throw new IOException("Session is null");
         }
 
-        // wrap the inner request in an outer request string
+        try (Throttler throttler = new Throttler()) {
-        String requestOuter = String.format(REQUEST_OUTER,
+            // start the session
-                jsonEscape(String.format(REQUEST_INNER, config.apiToken, jsonReplaceQuotes(requestJson))));
+            startSession();
 
-        // initialise the response
+            // session start failed
-        responseOuter = "";
+            Session session = this.session;
-        responsePending = true;
+            if (session == null) {
+                throw new IOException("Session is null");
+            }
 
-        IOException caughtException = null;
+            // wrap the inner request in an outer request string
-        throttle();
+            String requestOuter = String.format(REQUEST_OUTER,
-        try {
+                    jsonEscape(String.format(REQUEST_INNER, config.apiToken, jsonReplaceQuotes(requestJson))));
-            // send the request
-            logger.debug("hub '{}' sending characters:{}", hubId, requestOuter.length());
-            session.getRemote().sendString(requestOuter);
-            logger.trace("hub '{}' sent:{}", hubId, requestOuter);
 
-            // sleep and loop until we get a response, the socket is closed, or it times out
+            IOException caughtException = null;
-            Instant timeout = Instant.now().plusSeconds(config.socketTimeout);
+            try {
-            while (responsePending) {
+                // send the request
-                try {
+                logger.debug("hub '{}' sending characters:{}", hubId, requestOuter.length());
-                    Thread.sleep(SLEEP_MILLISECONDS);
+                session.getRemote().sendString(requestOuter);
-                    if (Instant.now().isAfter(timeout)) {
+                logger.trace("hub '{}' sent:{}", hubId, requestOuter);
-                        throw new IOException("Read timed out");
+
+                // sleep and loop until we get a response, the socket is closed, or it times out
+                Instant timeout = Instant.now().plusSeconds(config.socketTimeout);
+                while (!closing && websocketException == null && responses.isEmpty()) {
+                    try {
+                        Thread.sleep(SLEEP_MILLISECONDS);
+                        if (Instant.now().isAfter(timeout)) {
+                            throw new IOException("Read timed out");
+                        }
+                    } catch (InterruptedException e) {
+                        throw new IOException("Read interrupted", e);
                     }
-                } catch (InterruptedException e) {
-                    throw new IOException("Read interrupted", e);
                 }
+            } catch (IOException e) {
+                caughtException = e;
             }
-        } catch (IOException e) {
-            caughtException = e;
-        }
 
-        caughtException = caughtException != null ? caughtException
+            String responseOuter = !responses.isEmpty() ? responses.remove(0) : "";
-                : this.session == null ? new IOException("WebSocket session closed") : null;
+            logger.debug("hub '{}' received characters:{}", hubId, responseOuter.length());
+            logger.trace("hub '{}' received:{}", hubId, responseOuter);
 
-        logger.debug("hub '{}' received characters:{}", hubId, responseOuter.length());
+            // don't throw an exception if already closing
-        logger.trace("hub '{}' received:{}", hubId, responseOuter);
+            if (closing) {
+                return "{}";
+            }
 
-        // if an IOException was caught above, re-throw it again
+            // if an IOException was caught above, re-throw it again
-        if (caughtException != null) {
+            caughtException = websocketException != null ? websocketException : caughtException;
-            throw caughtException;
+            if (caughtException != null) {
-        }
+                throw caughtException;
+            }
 
-        try {
+            try {
-            Response responseDto = gson.fromJson(responseOuter, Response.class);
+                Response responseDto = gson.fromJson(responseOuter, Response.class);
-            if (responseDto == null) {
+                if (responseDto == null) {
-                throw new JsonSyntaxException("Response DTO is invalid");
+                    throw new JsonSyntaxException("Response DTO is invalid");
+                }
+                if (!NeoHubBindingConstants.HM_SET_COMMAND_RESPONSE.equals(responseDto.message_type)) {
+                    throw new JsonSyntaxException("DTO 'message_type' field is invalid");
+                }
+                String responseJson = responseDto.response;
+                if (responseJson == null) {
+                    throw new JsonSyntaxException("DTO 'response' field is null");
+                }
+                responseJson = jsonUnEscape(responseJson).strip();
+                if (!JsonParser.parseString(responseJson).isJsonObject()) {
+                    throw new JsonSyntaxException("DTO 'response' field is not a JSON object");
+                }
+                return responseJson;
+            } catch (JsonSyntaxException e) {
+                logger.debug("hub '{}' {}; response:{}", hubId, e.getMessage(), responseOuter);
+                throw new NeoHubException("Invalid response");
             }
-            if (!NeoHubBindingConstants.HM_SET_COMMAND_RESPONSE.equals(responseDto.message_type)) {
+        } catch (InterruptedException e) {
-                throw new JsonSyntaxException("DTO 'message_type' field is invalid");
+            throw new NeoHubException("Throttler was interrupted unexpectedly");
-            }
-            String responseJson = responseDto.response;
-            if (responseJson == null) {
-                throw new JsonSyntaxException("DTO 'response' field is null");
-            }
-            responseJson = jsonUnEscape(responseJson).strip();
-            if (!JsonParser.parseString(responseJson).isJsonObject()) {
-                throw new JsonSyntaxException("DTO 'response' field is not a JSON object");
-            }
-            return responseJson;
-        } catch (JsonSyntaxException e) {
-            logger.debug("hub '{}' {}; response:{}", hubId, e.getMessage(), responseOuter);
-            throw new NeoHubException("Invalid response");
         }
     }
 
     @Override
     public void close() {
-        closeSession();
+        closing = true;
+        Session session = this.session;
+        if (session != null) {
+            session.close();
+            this.session = null;
+        }
+        try {
+            trustManagerRegistration.unregister();
+        } catch (Exception e) {
+        }
         try {
             webSocketClient.stop();
         } catch (Exception e) {
@@ -242,25 +257,29 @@ public class NeoHubWebSocket extends NeoHubSocketBase {
     @OnWebSocketConnect
     public void onConnect(Session session) {
         logger.debug("hub '{}' onConnect() ok", hubId);
-        this.session = session;
     }
 
     @OnWebSocketClose
     public void onClose(int statusCode, String reason) {
-        logger.debug("hub '{}' onClose() statusCode:{}, reason:{}", hubId, statusCode, reason);
+        String closeMessage = String.format("onClose() statusCode:%d, reason:%s", statusCode, reason);
-        responsePending = false;
+        logger.debug("hub '{}' {}", hubId, closeMessage);
-        this.session = null;
+        websocketException = new IOException(closeMessage);
     }
 
     @OnWebSocketError
-    public void onError(Throwable cause) {
+    public void onError(@Nullable Throwable cause) {
-        logger.debug("hub '{}' onError() cause:{}", hubId, cause.getMessage());
+        logger.debug("hub '{}' onError() cause:{}", hubId, cause != null ? cause.getMessage() : "null");
-        closeSession();
+        websocketException = cause instanceof IOException ioCause ? ioCause : new IOException(cause);
     }
 
     @OnWebSocketMessage
-    public void onMessage(String msg) {
+    public synchronized void onMessage(String msg) {
-        responseOuter = msg.strip();
+        int responseCount = responses.size();
-        responsePending = false;
+        if (responseCount > 0) {
+            String errorMessage = String.format("onMessage() too many responses:%d", responseCount);
+            logger.debug("hub '{}' {}", hubId, errorMessage);
+            websocketException = new IOException(errorMessage);
+        }
+        responses.add(msg.strip());
     }
 }