Merge pull request #41856 from mkouba/ws-next-broadcasting-fix

WebSockets Next: broadcasting fixes
quarkusio · Jul 12, 2024 · 72af1d9 · 72af1d9
2 parents 29a8cf8 + cf67cc1
commit 72af1d9
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 5 deletions.
diff --git a/...s/websockets-next/runtime/src/main/java/io/quarkus/websockets/next/runtime/Endpoints.java b/...s/websockets-next/runtime/src/main/java/io/quarkus/websockets/next/runtime/Endpoints.java
@@ -291,7 +291,7 @@ private static boolean isSecurityFailure(Throwable throwable) {
                 || throwable instanceof ForbiddenException;
     }
 
-    private static boolean isWebSocketIsClosedFailure(Throwable throwable, WebSocketConnectionBase connection) {
+    static boolean isWebSocketIsClosedFailure(Throwable throwable, WebSocketConnectionBase connection) {
         if (!connection.isClosed()) {
             return false;
         }

diff --git a/...ext/runtime/src/main/java/io/quarkus/websockets/next/runtime/WebSocketConnectionImpl.java b/...ext/runtime/src/main/java/io/quarkus/websockets/next/runtime/WebSocketConnectionImpl.java
@@ -219,18 +219,26 @@ public Uni<Void> sendPong(Buffer data) {
             throw new UnsupportedOperationException();
         }
 
-        private <M> Uni<Void> doSend(BiFunction<WebSocketConnection, M, Uni<Void>> function, M message) {
+        private <M> Uni<Void> doSend(BiFunction<WebSocketConnection, M, Uni<Void>> sendFunction, M message) {
             Set<WebSocketConnection> connections = connectionManager.getConnections(generatedEndpointClass);
             if (connections.isEmpty()) {
                 return Uni.createFrom().voidItem();
             }
             List<Uni<Void>> unis = new ArrayList<>(connections.size());
             for (WebSocketConnection connection : connections) {
-                if (connection.isOpen() && (filter == null || filter.test(connection))) {
-                    unis.add(function.apply(connection, message));
+                if (connection.isOpen()
+                        && (filter == null || filter.test(connection))) {
+                    unis.add(sendFunction.apply(connection, message)
+                            // Intentionally ignore 'WebSocket is closed' failures
+                            // It might happen that the connection is closed in the mean time
+                            .onFailure(t -> Endpoints.isWebSocketIsClosedFailure(t, (WebSocketConnectionBase) connection))
+                            .recoverWithNull());
                 }
             }
-            return unis.isEmpty() ? Uni.createFrom().voidItem() : Uni.join().all(unis).andFailFast().replaceWithVoid();
+            if (unis.isEmpty()) {
+                return Uni.createFrom().voidItem();
+            }
+            return Uni.join().all(unis).andCollectFailures().replaceWithVoid();
         }
 
     }