From 76df1108d745b358bae713ea06d130b6c64c9386 Mon Sep 17 00:00:00 2001
From: Redkale <22250530@qq.com>
Date: Sat, 1 Jul 2017 15:38:47 +0800
Subject: [PATCH] =?UTF-8?q?WebSocketEngine.broadcastMessage=20=E5=A2=9E?=
 =?UTF-8?q?=E5=8A=A0=20Predicate<WebSocket>=20=E5=8F=82=E6=95=B0?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 src/org/redkale/net/http/WebSocketEngine.java | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/src/org/redkale/net/http/WebSocketEngine.java b/src/org/redkale/net/http/WebSocketEngine.java
index 07e8787b3..1c107f7ec 100644
--- a/src/org/redkale/net/http/WebSocketEngine.java
+++ b/src/org/redkale/net/http/WebSocketEngine.java
@@ -10,6 +10,7 @@ import java.io.*;
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.*;
+import java.util.function.Predicate;
 import java.util.logging.*;
 import java.util.stream.*;
 import org.redkale.convert.Convert;
@@ -128,6 +129,10 @@ public final class WebSocketEngine {
     }
 
     public CompletableFuture<Integer> broadcastMessage(final Object message, final boolean last) {
+        return broadcastMessage(null, message, last);
+    }
+
+    public CompletableFuture<Integer> broadcastMessage(final Predicate<WebSocket> predicate, final Object message, final boolean last) {
         if (message instanceof CompletableFuture) {
             return ((CompletableFuture) message).thenCompose((json) -> broadcastMessage(json, last));
         }
@@ -140,11 +145,13 @@ public final class WebSocketEngine {
             CompletableFuture<Integer> future = null;
             if (single) {
                 for (WebSocket websocket : websockets.values()) {
+                    if (predicate != null && !predicate.test(websocket)) continue;
                     future = future == null ? websocket.sendPacket(packet) : future.thenCombine(websocket.sendPacket(packet), (a, b) -> a | (Integer) b);
                 }
             } else {
                 for (List<WebSocket> list : websockets2.values()) {
                     for (WebSocket websocket : list) {
+                        if (predicate != null && !predicate.test(websocket)) continue;
                         future = future == null ? websocket.sendPacket(packet) : future.thenCombine(websocket.sendPacket(packet), (a, b) -> a | (Integer) b);
                     }
                 }