[camel] branch camel-3.11.x updated: CAMEL-17503: fix websocket client unable to reconnect to Server after server reboot (#6792)

[davsclaus]

This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a commit to branch camel-3.11.x
in repository https://gitbox.apache.org/repos/asf/camel.git


The following commit(s) were added to refs/heads/camel-3.11.x by this push:
     new a795221  CAMEL-17503: fix websocket client unable to reconnect to 
Server after server reboot (#6792)
a795221 is described below

commit a795221c591780830d00e5e2dbf8cfcc43c662f9
Author: CG <cg...@users.noreply.github.com>
AuthorDate: Thu Jan 20 22:50:13 2022 +0800

    CAMEL-17503: fix websocket client unable to reconnect to Server after 
server reboot (#6792)
    
    Co-authored-by: CG <chengguan.po...@thalesdigital.io>
---
 .../apache/camel/component/ahc/ws/WsEndpoint.java  |  8 ++++++-
 .../apache/camel/component/ahc/ws/TestServlet.java |  1 +
 .../component/ahc/ws/WsProducerConsumerTest.java   | 27 +++++++++++++++++++++-
 3 files changed, 34 insertions(+), 2 deletions(-)

diff --git 
a/components/camel-ahc-ws/src/main/java/org/apache/camel/component/ahc/ws/WsEndpoint.java
 
b/components/camel-ahc-ws/src/main/java/org/apache/camel/component/ahc/ws/WsEndpoint.java
index 2dbce5d..bc72c41 100644
--- 
a/components/camel-ahc-ws/src/main/java/org/apache/camel/component/ahc/ws/WsEndpoint.java
+++ 
b/components/camel-ahc-ws/src/main/java/org/apache/camel/component/ahc/ws/WsEndpoint.java
@@ -172,9 +172,15 @@ public class WsEndpoint extends AhcEndpoint {
         }
 
         @Override
-        public void onClose(WebSocket websocket, int code, String reason) {
+        public void onClose(WebSocket socket, int code, String reason) {
             LOG.debug("websocket closed - reconnecting");
             try {
+                if (websocket != null) {
+                    // set websocket to null and remove the listener
+                    websocket.removeWebSocketListener(listener);
+                    websocket.sendCloseFrame();
+                    websocket = null;
+                }
                 reConnect();
             } catch (Exception e) {
                 LOG.warn("Error re-connecting to websocket", e);
diff --git 
a/components/camel-ahc-ws/src/test/java/org/apache/camel/component/ahc/ws/TestServlet.java
 
b/components/camel-ahc-ws/src/test/java/org/apache/camel/component/ahc/ws/TestServlet.java
index 440b9d1..99a924b 100644
--- 
a/components/camel-ahc-ws/src/test/java/org/apache/camel/component/ahc/ws/TestServlet.java
+++ 
b/components/camel-ahc-ws/src/test/java/org/apache/camel/component/ahc/ws/TestServlet.java
@@ -34,6 +34,7 @@ public class TestServlet {
     @OnWebSocketConnect
     public void handleConnect(Session session) {
         this.session = session;
+        send("Connected!");
     }
 
     @OnWebSocketMessage
diff --git 
a/components/camel-ahc-ws/src/test/java/org/apache/camel/component/ahc/ws/WsProducerConsumerTest.java
 
b/components/camel-ahc-ws/src/test/java/org/apache/camel/component/ahc/ws/WsProducerConsumerTest.java
index 1f8a800..3711794 100644
--- 
a/components/camel-ahc-ws/src/test/java/org/apache/camel/component/ahc/ws/WsProducerConsumerTest.java
+++ 
b/components/camel-ahc-ws/src/test/java/org/apache/camel/component/ahc/ws/WsProducerConsumerTest.java
@@ -40,6 +40,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
 public class WsProducerConsumerTest extends CamelTestSupport {
 
     protected static final String TEST_MESSAGE = "Hello World!";
+    protected static final String TEST_CONNECTED_MESSAGE = "Connected!";
     protected static final int PORT = AvailablePortFinder.getNextAvailable();
 
     private static final Logger LOG = 
LoggerFactory.getLogger(WsProducerConsumerTest.class);
@@ -139,9 +140,27 @@ public class WsProducerConsumerTest extends 
CamelTestSupport {
         mock.assertIsSatisfied();
     }
 
+    @Test
+    public void testRestartServer() throws Exception {
+        MockEndpoint mock = getMockEndpoint("mock:restart-result");
+        mock.expectedBodiesReceived(TEST_CONNECTED_MESSAGE);
+
+        mock.assertIsSatisfied();
+
+        LOG.info("Restarting Test Server");
+        stopTestServer();
+        startTestServer();
+
+        resetMocks();
+
+        mock.expectedBodiesReceived(TEST_CONNECTED_MESSAGE);
+
+        mock.assertIsSatisfied();
+    }
+
     @Override
     protected RouteBuilder[] createRouteBuilders() throws Exception {
-        RouteBuilder[] rbs = new RouteBuilder[2];
+        RouteBuilder[] rbs = new RouteBuilder[3];
         rbs[0] = new RouteBuilder() {
             public void configure() {
                 from("direct:input").routeId("foo")
@@ -154,6 +173,12 @@ public class WsProducerConsumerTest extends 
CamelTestSupport {
                         .to("mock:result");
             }
         };
+        rbs[2] = new RouteBuilder() {
+            public void configure() {
+                from("ahc-ws://localhost:" + PORT + 
"/restart").routeId("restart")
+                        .to("mock:restart-result");
+            }
+        };
         return rbs;
     }
 }