Merge pull request #1343 from gtroitsk/feature/websockets/number-of-o…

…pened-sessions

Add client session single opening assert to WebSockets tests

websockets/quarkus-websockets/src/test/java/io/quarkus/ts/websockets/producer/WebSocketsProducerConsumerIT.java

@@ -5,10 +5,9 @@
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
-import java.util.Set;
-import java.util.concurrent.ConcurrentSkipListSet;
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import jakarta.websocket.ClientEndpoint;
 import jakarta.websocket.ContainerProvider;
@@ -42,6 +41,8 @@ public void smoke() throws Exception {
             session.getAsyncRemote().sendText("hello world");
             assertMessage(">> stu: hello world", client);
         }
+
+        assertSessionWasOpenedJustOnce(client);
     }
 
     @Test
@@ -59,6 +60,8 @@ public void chatting() throws Exception {
                 assertMessage(">> bob: hello alice", bobChat, aliceChat);
             }
         }
+
+        assertSessionWasOpenedJustOnce(aliceChat, bobChat);
     }
 
     @Test
@@ -76,6 +79,8 @@ public void chattingSync() throws Exception {
                 assertMessage(">> bob: hello alice", bobChat, aliceChat);
             }
         }
+
+        assertSessionWasOpenedJustOnce(aliceChat, bobChat);
     }
 
     @Test
@@ -97,6 +102,8 @@ public void trio() throws Exception {
                 assertMessage("User athos left", porthosChat, aramisChat);
             }
         }
+
+        assertSessionWasOpenedJustOnce(athosChat, porthosChat, aramisChat);
     }
 
     @Test
@@ -106,6 +113,8 @@ public void i18n() throws Exception {
             session.getAsyncRemote().sendText("今日は přátelé, как дела? \uD83E\uDED6 ?");
             assertMessage(">> traveler: 今日は přátelé, как дела? \uD83E\uDED6 ?", client);
         }
+
+        assertSessionWasOpenedJustOnce(client);
     }
 
     @Test
@@ -117,6 +126,8 @@ public void push() throws Exception {
             assertMessage("Three", client);
             assertMessage("Four", client);
         }
+
+        assertSessionWasOpenedJustOnce(client);
     }
 
     private static URI getUri(String with) throws URISyntaxException {
@@ -140,6 +151,12 @@ private static void assertMessage(String expectedMessage, Client... clients) {
         }
     }
 
+    private static void assertSessionWasOpenedJustOnce(Client... clients) {
+        for (Client client : clients) {
+            Assertions.assertEquals(1, client.getNumberOfOpenedSessions());
+        }
+    }
+
     private void broadcastPlainMsgAsync(Session session, String msg) {
         session.getAsyncRemote().sendText(msg, result -> {
             if (result.getException() != null) {
@@ -161,7 +178,7 @@ private void broadcastPlainMsg(Session session, String msg) {
     @ClientEndpoint
     public static class Client {
         private final LinkedBlockingDeque<String> messages = new LinkedBlockingDeque<>();
-        private final Set<String> joins = new ConcurrentSkipListSet<>();
+        private final AtomicInteger sessionOpenCounter = new AtomicInteger();
 
         @OnOpen
         public void open(Session session) {
@@ -173,6 +190,8 @@ public void open(Session session) {
                     Assertions.fail(result.getException());
                 }
             });
+
+            sessionOpenCounter.incrementAndGet();
         }
 
         @OnMessage
@@ -185,5 +204,9 @@ void message(String msg) {
         public String getMessage() throws InterruptedException {
             return messages.poll(10, TimeUnit.SECONDS);
         }
+
+        public int getNumberOfOpenedSessions() {
+            return this.sessionOpenCounter.get();
+        }
     }
 }