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

WebSocket

Browse Source
This commit is contained in:
redkale
2023-10-09 09:28:21 +08:00
parent c58c603e6d
commit c66aafe7fe
1 changed files with 36 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
src/main/java/org/redkale/net/http/WebSocket.java
View File

@@ -11,7 +11,9 @@ import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets; import java.nio.charset.StandardCharsets;
import java.util.*; import java.util.*;
import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;
import java.util.logging.*; import java.util.logging.*;
import java.util.stream.Stream; import java.util.stream.Stream;
import java.util.zip.*; import java.util.zip.*;
@@ -917,13 +919,34 @@ public abstract class WebSocket<G extends Serializable, T> {
} }
} }
//closeRunner /**
* 会出现线程锁的问题
* io.vertx.core.VertxException: Thread blocked
* at java.base/jdk.internal.misc.Unsafe.park(Native Method)
* at java.base/java.util.concurrent.locks.LockSupport.park(LockSupport.java:221)
* at java.base/java.util.concurrent.CompletableFuture$Signaller.block(CompletableFuture.java:1864)
* at java.base/java.util.concurrent.ForkJoinPool.unmanagedBlock(ForkJoinPool.java:3780)
* at java.base/java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3725)
* at java.base/java.util.concurrent.CompletableFuture.waitingGet(CompletableFuture.java:1898)
* at java.base/java.util.concurrent.CompletableFuture.join(CompletableFuture.java:2117)
* at org.redkale.net.http.WebSocket.close(WebSocket.java:916)
* at org.redkale.net.http.WebSocketEngine.forceCloseLocalWebSocket(WebSocketEngine.java:229)
* at org.redkale.net.http.WebSocketNode.forceCloseWebSocket(WebSocketNode.java:428)
* at org.redkale.net.http.WebSocketNode.forceCloseWebSocket(WebSocketNode.java:410)
* at org.redkale.net.http.WebSocket.forceCloseWebSocket(WebSocket.java:575)
* at org.redkale.net.http.WebSocket.onSingleRepeatConnect(WebSocket.java:856)
* at org.redkale.net.http.WebSocketServlet$1.lambda$completed$2(WebSocketServlet.java:314)
* at java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863)
* at java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841)
* at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)
* at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2179)
*/
CompletableFuture<Void> kill(int code, String reason) { CompletableFuture<Void> kill(int code, String reason) {
if (closed.compareAndSet(false, true)) { if (closed.compareAndSet(false, true)) {
if (_channel == null) { if (_channel == null) {
return null; return null;
} }
CompletableFuture<Void> future = _engine.removeLocalThenDisconnect(this); Supplier<CompletableFuture> compose = () -> {
_channel.dispose(); _channel.dispose();
if (_readHandler != null) { if (_readHandler != null) {
_readHandler.byteArrayPool.accept(_readHandler.halfFrameBytes); _readHandler.byteArrayPool.accept(_readHandler.halfFrameBytes);
@@ -931,11 +954,11 @@ public abstract class WebSocket<G extends Serializable, T> {
if (_writeHandler != null) { if (_writeHandler != null) {
_writeHandler.byteArrayPool.accept(_writeHandler.writeArray); _writeHandler.byteArrayPool.accept(_writeHandler.writeArray);
} }
CompletableFuture closeFuture = onClose(code, reason); return onClose(code, reason);
if (closeFuture == null) { };
return future; CompletableFuture<Void> future = _engine.removeLocalThenDisconnect(this);
} return future == null ? compose.get()
return CompletableFuture.allOf(future, closeFuture); : future.exceptionally(t -> null).thenCompose(v -> (CompletionStage) compose);
} else { } else {
return null; return null;
} }