Make connect and disconnect thread safe

Author: georgweiss

core/websocket/src/main/java/org/phoebus/core/websocket/springframework/WebSocketClientService.java

@@ -46,6 +46,10 @@
  *         <li>Publish a topic named /path/web-socket/messages, where path is optional.</li>
  *     </ul>
  * </p>
+ * <p>
+ *     <b>NOTE:</b> client code <i>must</i> call the {@link #shutdown()} method to not
+ *     leave the web socket connection alive.
+ * </p>
  */
 public class WebSocketClientService {
 
@@ -122,26 +126,18 @@ public void sendEcho(String message) {
         if (stompSession != null && stompSession.isConnected() && echoEndpoint != null) {
             stompSession.send(echoEndpoint, message);
         }
-
     }
 
     /**
-     * Disconnects the STOMP session if non-null and connected.
+     * Disconnects the socket if connected and terminates connection thread.
      */
-    public void disconnect() {
+    public synchronized void shutdown() {
+        attemptReconnect.set(false);
         if (stompSession != null && stompSession.isConnected()) {
             stompSession.disconnect();
         }
     }
 
-    /**
-     * Disconnects the socket if connected and terminates connection thread.
-     */
-    public void shutdown() {
-        attemptReconnect.set(false);
-        disconnect();
-    }
-
     /**
      * Attempts to connect to the remote peer, both in initial connection and in a reconnection scenario.
      * If connection fails, new attempts are made every 10s until successful.
@@ -156,23 +152,25 @@ public void connect() {
         stompClient.setTaskScheduler(threadPoolTaskScheduler);
         stompClient.setDefaultHeartbeat(new long[]{30000, 30000});
         StompSessionHandler sessionHandler = new StompSessionHandler();
+        logger.log(Level.INFO, "Attempting web socket connection to " + connectUrl);
         new Thread(() -> {
             while (attemptReconnect.get()) {
-                logger.log(Level.INFO, "Attempting web socket connection to " + connectUrl);
                 try {
-                    stompSession = stompClient.connect(connectUrl, sessionHandler).get();
-                    stompSession.subscribe(this.subscriptionEndpoint, new StompFrameHandler() {
-                        @Override
-                        public Type getPayloadType(StompHeaders headers) {
-                            return String.class;
-                        }
+                    synchronized (WebSocketClientService.this) {
+                        stompSession = stompClient.connect(connectUrl, sessionHandler).get();
+                        stompSession.subscribe(this.subscriptionEndpoint, new StompFrameHandler() {
+                            @Override
+                            public Type getPayloadType(StompHeaders headers) {
+                                return String.class;
+                            }
 
-                        @Override
-                        public void handleFrame(StompHeaders headers, Object payload) {
-                            logger.log(Level.INFO, "Handling subscription frame: " + payload);
-                            webSocketMessageHandlers.forEach(h -> h.handleWebSocketMessage((String) payload));
-                        }
-                    });
+                            @Override
+                            public void handleFrame(StompHeaders headers, Object payload) {
+                                logger.log(Level.INFO, "Handling subscription frame: " + payload);
+                                webSocketMessageHandlers.forEach(h -> h.handleWebSocketMessage((String) payload));
+                            }
+                        });
+                    }
                 } catch (Exception e) {
                     logger.log(Level.WARNING, "Got exception when trying to connect", e);
                 }