lxy/redkale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

WebSocketEngine.broadcastMessage 增加 Predicate<WebSocket> 参数

Browse Source
This commit is contained in:
Redkale
2017-07-01 15:38:47 +08:00
parent 941d09cde2
commit 76df1108d7
1 changed files with 7 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/org/redkale/net/http/WebSocketEngine.java
View File

@@ -10,6 +10,7 @@ import java.io.*;
import java.util.*; import java.util.*;
import java.util.concurrent.*; import java.util.concurrent.*;
import java.util.concurrent.atomic.*; import java.util.concurrent.atomic.*;
import java.util.function.Predicate;
import java.util.logging.*; import java.util.logging.*;
import java.util.stream.*; import java.util.stream.*;
import org.redkale.convert.Convert; import org.redkale.convert.Convert;
@@ -128,6 +129,10 @@ public final class WebSocketEngine {
} }
public CompletableFuture<Integer> broadcastMessage(final Object message, final boolean last) { 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) { if (message instanceof CompletableFuture) {
return ((CompletableFuture) message).thenCompose((json) -> broadcastMessage(json, last)); return ((CompletableFuture) message).thenCompose((json) -> broadcastMessage(json, last));
} }
@@ -140,11 +145,13 @@ public final class WebSocketEngine {
CompletableFuture<Integer> future = null; CompletableFuture<Integer> future = null;
if (single) { if (single) {
for (WebSocket websocket : websockets.values()) { 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); future = future == null ? websocket.sendPacket(packet) : future.thenCombine(websocket.sendPacket(packet), (a, b) -> a | (Integer) b);
} }
} else { } else {
for (List<WebSocket> list : websockets2.values()) { for (List<WebSocket> list : websockets2.values()) {
for (WebSocket websocket : list) { 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); future = future == null ? websocket.sendPacket(packet) : future.thenCombine(websocket.sendPacket(packet), (a, b) -> a | (Integer) b);
} }
} }