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

sendToChannel0

Browse Source
This commit is contained in:
redkale
2024-10-21 20:41:57 +08:00
parent 72fe0dc2ba
commit 1d9f385da9
2 changed files with 5 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -273,23 +273,13 @@ public abstract class WebSocket<G extends Serializable, T> {
* @return 0表示成功， 非0表示错误码 * @return 0表示成功， 非0表示错误码
*/ */
CompletableFuture<Integer> sendPacket(WebSocketPacket packet) { CompletableFuture<Integer> sendPacket(WebSocketPacket packet) {
if (this._readHandler == null) { // if (this._writeIOThread == null) { if (this._readHandler == null) {
if (delayPackets == null) { if (delayPackets == null) {
delayPackets = new ArrayList<>(); delayPackets = new ArrayList<>();
} }
delayPackets.add(packet); delayPackets.add(packet);
return CompletableFuture.completedFuture(RETCODE_DELAYSEND); return CompletableFuture.completedFuture(RETCODE_DELAYSEND);
} }
return _sendToChannel(packet);
}
/**
* 给自身发送消息体, 包含二进制/文本
*
* @param packet WebSocketPacket
* @return 0表示成功， 非0表示错误码
*/
CompletableFuture<Integer> _sendToChannel(WebSocketPacket packet) {
if (_channel == null || closed.get()) { if (_channel == null || closed.get()) {
return CompletableFuture.completedFuture(RETCODE_WSOCKET_CLOSED); return CompletableFuture.completedFuture(RETCODE_WSOCKET_CLOSED);
} }
@@ -307,7 +297,7 @@ public abstract class WebSocket<G extends Serializable, T> {
* @param packets WebSocketPacket集合 * @param packets WebSocketPacket集合
* @return 0表示成功， 非0表示错误码 * @return 0表示成功， 非0表示错误码
*/ */
CompletableFuture<Integer> _sendToChannel(List<WebSocketPacket> packets) { CompletableFuture<Integer> sendToChannel0(List<WebSocketPacket> packets) {
if (_channel == null || closed.get()) { if (_channel == null || closed.get()) {
return CompletableFuture.completedFuture(RETCODE_WSOCKET_CLOSED); return CompletableFuture.completedFuture(RETCODE_WSOCKET_CLOSED);
} }

10
src/main/java/org/redkale/net/http/WebSocketServlet.java
View File

@@ -80,7 +80,7 @@ public abstract class WebSocketServlet extends HttpServlet implements Resourcabl
private final BiConsumer<WebSocket, Object> restMessageConsumer = createRestOnMessageConsumer(); private final BiConsumer<WebSocket, Object> restMessageConsumer = createRestOnMessageConsumer();
private final ObjectPool<ByteArray> byteArrayPool = private final ObjectPool<ByteArray> byteArrayPool =
ObjectPool.createSafePool(1000, () -> new ByteArray(), null, ByteArray::recycle); ObjectPool.createSafePool(1024, ps -> new ByteArray(), null, ByteArray::recycle);
// RestWebSocket时会被修改 // RestWebSocket时会被修改
protected Type messageRestType; protected Type messageRestType;
@@ -411,9 +411,7 @@ public abstract class WebSocketServlet extends HttpServlet implements Resourcabl
if (webSocket.delayPackets != null) { // 存在待发送的消息 if (webSocket.delayPackets != null) { // 存在待发送的消息
List<WebSocketPacket> delayPackets = webSocket.delayPackets; List<WebSocketPacket> delayPackets = webSocket.delayPackets;
webSocket.delayPackets = null; webSocket.delayPackets = null;
// CompletableFuture<Integer> cf = webSocket._writeIOThread.send(webSocket, CompletableFuture<Integer> cf = webSocket.sendToChannel0(delayPackets);
// delayPackets.toArray(new WebSocketPacket[delayPackets.size()]));
CompletableFuture<Integer> cf = webSocket._sendToChannel(delayPackets);
cf.whenComplete((Integer v, Throwable t) -> { cf.whenComplete((Integer v, Throwable t) -> {
Traces.currentTraceid(request.getTraceid()); Traces.currentTraceid(request.getTraceid());
if (userid == null || t != null) { if (userid == null || t != null) {
@@ -437,9 +435,7 @@ public abstract class WebSocketServlet extends HttpServlet implements Resourcabl
if (webSocket.delayPackets != null) { // 存在待发送的消息 if (webSocket.delayPackets != null) { // 存在待发送的消息
List<WebSocketPacket> delayPackets = webSocket.delayPackets; List<WebSocketPacket> delayPackets = webSocket.delayPackets;
webSocket.delayPackets = null; webSocket.delayPackets = null;
// CompletableFuture<Integer> cf = webSocket._writeIOThread.send(webSocket, CompletableFuture<Integer> cf = webSocket.sendToChannel0(delayPackets);
// delayPackets.toArray(new WebSocketPacket[delayPackets.size()]));
CompletableFuture<Integer> cf = webSocket._sendToChannel(delayPackets);
cf.whenComplete((Integer v, Throwable t) -> { cf.whenComplete((Integer v, Throwable t) -> {
Traces.currentTraceid(request.getTraceid()); Traces.currentTraceid(request.getTraceid());
if (sessionid == null || t != null) { if (sessionid == null || t != null) {