From c961f5f80c980945be653eeae7df441a9bd7e709 Mon Sep 17 00:00:00 2001 From: wentch <22250530@qq.com> Date: Wed, 23 Dec 2015 16:48:29 +0800 Subject: [PATCH] --- src/org/redkale/net/http/WebSocket.java | 72 +++++++++---------- src/org/redkale/net/http/WebSocketEngine.java | 24 +++++-- src/org/redkale/net/http/WebSocketGroup.java | 2 +- src/org/redkale/net/http/WebSocketRunner.java | 52 +++++++------- .../redkale/net/http/WebSocketServlet.java | 9 ++- 5 files changed, 81 insertions(+), 78 deletions(-) diff --git a/src/org/redkale/net/http/WebSocket.java b/src/org/redkale/net/http/WebSocket.java index 4af3357aa..b53fbd129 100644 --- a/src/org/redkale/net/http/WebSocket.java +++ b/src/org/redkale/net/http/WebSocket.java @@ -13,21 +13,21 @@ import org.redkale.net.*; /** * 一个WebSocket连接对应一个WebSocket实体，即一个WebSocket会绑定一个TCP连接。 - * WebSocket 有两种模式: - * 1) 普通模式: 协议上符合HTML5规范, 其流程顺序如下: - * 1.1 onOpen 如果方法返回null，则视为WebSocket的连接不合法，框架会强制关闭WebSocket连接；通常用于判断登录态。 - * 1.2 createGroupid 如果方法返回null，则视为WebSocket的连接不合法，框架会强制关闭WebSocket连接；通常用于判断用户权限是否符合。 - * 1.3 onConnected WebSocket成功连接后在准备接收数据前回调此方法。 - * 1.4 onMessage/onFragment+ WebSocket接收到消息后回调此消息类方法。 - * 1.5 onClose WebSocket被关闭后回调此方法。 - * + * WebSocket 有两种模式: + * 1) 普通模式: 协议上符合HTML5规范, 其流程顺序如下: + * 1.1 onOpen 如果方法返回null，则视为WebSocket的连接不合法，框架会强制关闭WebSocket连接；通常用于判断登录态。 + * 1.2 createGroupid 如果方法返回null，则视为WebSocket的连接不合法，框架会强制关闭WebSocket连接；通常用于判断用户权限是否符合。 + * 1.3 onConnected WebSocket成功连接后在准备接收数据前回调此方法。 + * 1.4 onMessage/onFragment+ WebSocket接收到消息后回调此消息类方法。 + * 1.5 onClose WebSocket被关闭后回调此方法。 + * * 此模式下 以上方法都应该被重载。 - * - * 2) 原始二进制模式: 此模式有别于HTML5规范，可以视为原始的TCP连接。通常用于音频视频通讯场景。期流程顺序如下: - * 2.1 onOpen 如果方法返回null，则视为WebSocket的连接不合法，框架会强制关闭WebSocket连接；通常用于判断登录态。 - * 2.2 createGroupid 如果方法返回null，则视为WebSocket的连接不合法，框架会强制关闭WebSocket连接；通常用于判断用户权限是否符合。 - * 2.3 onRead WebSocket成功连接后回调此方法， 由此方法处理原始的TCP连接， 同时业务代码去控制WebSocket的关闭。 - * + * + * 2) 原始二进制模式: 此模式有别于HTML5规范，可以视为原始的TCP连接。通常用于音频视频通讯场景。期流程顺序如下: + * 2.1 onOpen 如果方法返回null，则视为WebSocket的连接不合法，框架会强制关闭WebSocket连接；通常用于判断登录态。 + * 2.2 createGroupid 如果方法返回null，则视为WebSocket的连接不合法，框架会强制关闭WebSocket连接；通常用于判断用户权限是否符合。 + * 2.3 onRead WebSocket成功连接后回调此方法， 由此方法处理原始的TCP连接， 同时业务代码去控制WebSocket的关闭。 + * * 此模式下 以上方法都应该被重载。 *

* @@ -56,17 +56,15 @@ public abstract class WebSocket { public static final int RETCODE_WSOFFLINE = 1 << 8; //256 - WebSocketRunner runner; + WebSocketRunner runner; //不可能为空 - WebSocketEngine engine; + WebSocketEngine _engine; //不可能为空 - WebSocketGroup group; + WebSocketGroup _group; //不可能为空 - WebSocketNode node; + Serializable _sessionid; //不可能为空 - Serializable sessionid; - - Serializable groupid; + Serializable _groupid; //不可能为空 private final long createtime = System.currentTimeMillis(); @@ -80,7 +78,7 @@ public abstract class WebSocket { * 发送消息体, 包含二进制/文本 *

* @param packet - * @return + * @return */ public final int send(WebSocketPacket packet) { if (this.runner != null) return this.runner.sendMessage(packet); @@ -98,7 +96,7 @@ public abstract class WebSocket { * 发送单一的文本消息 *

* @param text 不可为空 - * @return + * @return */ public final int send(String text) { return send(text, true); @@ -109,7 +107,7 @@ public abstract class WebSocket { *

* @param text 不可为空 * @param last 是否最后一条 - * @return + * @return */ public final int send(String text, boolean last) { return send(new WebSocketPacket(text, last)); @@ -119,7 +117,7 @@ public abstract class WebSocket { * 发送单一的二进制消息 *

* @param data - * @return + * @return */ public final int send(byte[] data) { return send(data, true); @@ -142,7 +140,7 @@ public abstract class WebSocket { *

* @param data 不可为空 * @param last 是否最后一条 - * @return + * @return */ public final int send(byte[] data, boolean last) { return send(new WebSocketPacket(data, last)); @@ -152,8 +150,8 @@ public abstract class WebSocket { * 发送消息, 消息类型是String或byte[] *

* @param message 不可为空, 只能是String或者byte[] - * @param last 是否最后一条 - * @return + * @param last 是否最后一条 + * @return */ public final int send(Serializable message, boolean last) { return send(new WebSocketPacket(message, last)); @@ -253,13 +251,13 @@ public abstract class WebSocket { } private int sendMessage(Serializable groupid, boolean recent, String text, boolean last) { - if (node == null) return RETCODE_NODESERVICE_NULL; - return node.sendMessage(groupid, recent, text, last); + if (_engine.node == null) return RETCODE_NODESERVICE_NULL; + return _engine.node.sendMessage(groupid, recent, text, last); } private int sendMessage(Serializable groupid, boolean recent, byte[] data, boolean last) { - if (node == null) return RETCODE_NODESERVICE_NULL; - return node.sendMessage(groupid, recent, data, last); + if (_engine.node == null) return RETCODE_NODESERVICE_NULL; + return _engine.node.sendMessage(groupid, recent, data, last); } /** @@ -301,7 +299,7 @@ public abstract class WebSocket { * @return */ public final Serializable getGroupid() { - return groupid; + return _groupid; } /** @@ -310,7 +308,7 @@ public abstract class WebSocket { * @return */ public final Serializable getSessionid() { - return sessionid; + return _sessionid; } //------------------------------------------------------------------- @@ -320,7 +318,7 @@ public abstract class WebSocket { * @return */ protected final WebSocketGroup getWebSocketGroup() { - return group; + return _group; } /** @@ -330,11 +328,11 @@ public abstract class WebSocket { * @return */ protected final WebSocketGroup getWebSocketGroup(Serializable groupid) { - return engine.getWebSocketGroup(groupid); + return _engine.getWebSocketGroup(groupid); } protected final Collection getWebSocketGroups() { - return engine.getWebSocketGroups(); + return _engine.getWebSocketGroups(); } //------------------------------------------------------------------- diff --git a/src/org/redkale/net/http/WebSocketEngine.java b/src/org/redkale/net/http/WebSocketEngine.java index 4a16c1e17..4d9509a51 100644 --- a/src/org/redkale/net/http/WebSocketEngine.java +++ b/src/org/redkale/net/http/WebSocketEngine.java @@ -27,6 +27,8 @@ public final class WebSocketEngine { private final String engineid; + protected final WebSocketNode node; + private final Map containers = new ConcurrentHashMap<>(); private ScheduledThreadPoolExecutor scheduler; @@ -35,8 +37,9 @@ public final class WebSocketEngine { protected final boolean finest; - protected WebSocketEngine(String engineid, Logger logger) { + protected WebSocketEngine(String engineid, WebSocketNode node, Logger logger) { this.engineid = engineid; + this.node = node; this.logger = logger; this.index = sequence.getAndIncrement(); this.finest = logger.isLoggable(Level.FINEST); @@ -60,19 +63,26 @@ public final class WebSocketEngine { } void add(WebSocket socket) { - WebSocketGroup group = containers.get(socket.groupid); + WebSocketGroup group = containers.get(socket._groupid); if (group == null) { - group = new WebSocketGroup(socket.groupid); - containers.put(socket.groupid, group); + group = new WebSocketGroup(socket._groupid); + containers.putIfAbsent(socket._groupid, group); } group.add(socket); + if (node != null) node.connect(socket._groupid, engineid); } void remove(WebSocket socket) { - WebSocketGroup group = containers.get(socket.groupid); - if (group == null) return; + final WebSocketGroup group = containers.get(socket._groupid); + if (group == null) { + if (node != null) node.disconnect(socket._groupid, engineid); + return; + } group.remove(socket); - if (group.isEmpty()) containers.remove(socket.groupid); + if (group.isEmpty()) { + containers.remove(socket._groupid); + if (node != null) node.disconnect(socket._groupid, engineid); + } } Collection getWebSocketGroups() { diff --git a/src/org/redkale/net/http/WebSocketGroup.java b/src/org/redkale/net/http/WebSocketGroup.java index 6ddb35814..8d00ef97e 100644 --- a/src/org/redkale/net/http/WebSocketGroup.java +++ b/src/org/redkale/net/http/WebSocketGroup.java @@ -42,7 +42,7 @@ public final class WebSocketGroup { } void add(WebSocket socket) { - socket.group = this; + socket._group = this; this.recentWebSocket = socket; list.add(socket); } diff --git a/src/org/redkale/net/http/WebSocketRunner.java b/src/org/redkale/net/http/WebSocketRunner.java index 27f4aa495..90842a2c4 100644 --- a/src/org/redkale/net/http/WebSocketRunner.java +++ b/src/org/redkale/net/http/WebSocketRunner.java @@ -48,7 +48,7 @@ public class WebSocketRunner implements Runnable { public WebSocketRunner(Context context, WebSocket webSocket, AsyncConnection channel, final boolean wsbinary) { this.context = context; - this.engine = webSocket.engine; + this.engine = webSocket._engine; this.webSocket = webSocket; this.channel = channel; this.wsbinary = wsbinary; @@ -63,7 +63,6 @@ public class WebSocketRunner implements Runnable { public void run() { final boolean debug = this.coder.debugable; try { - if (webSocket.node != null) webSocket.node.connect(webSocket.groupid, webSocket.engine.getEngineid()); webSocket.onConnected(); channel.setReadTimeoutSecond(300); //读取超时5分钟 if (channel.isOpen()) { @@ -119,7 +118,7 @@ public class WebSocketRunner implements Runnable { readBuffer.clear(); channel.read(readBuffer, null, this); } - webSocket.group.setRecentWebSocket(webSocket); + webSocket._group.setRecentWebSocket(webSocket); try { if (packet.type == FrameType.TEXT) { webSocket.onMessage(packet.getPayload()); @@ -244,10 +243,6 @@ public class WebSocketRunner implements Runnable { readBuffer = null; writeBuffer = null; engine.remove(webSocket); - if (webSocket.node != null) { - WebSocketGroup group = webSocket.getWebSocketGroup(); - if (group == null || group.isEmpty()) webSocket.node.disconnect(webSocket.groupid, webSocket.engine.getEngineid()); - } webSocket.onClose(0, null); } } @@ -370,27 +365,28 @@ public class WebSocketRunner implements Runnable { private Logger logger; /** - 0 1 2 3 - 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 - +-+-+-+-+-------+-+-------------+-------------------------------+ - |F|R|R|R| opcode|M| Payload len | Extended payload length | - |I|S|S|S| (4) |A| (7) | (16/64) | - |N|V|V|V| |S| | (if payload len==126/127) | - | |1|2|3| |K| | | - +-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - + - | Extended payload length continued, if payload len == 127 | - + - - - - - - - - - - - - - - - +-------------------------------+ - | |Masking-key, if MASK set to 1 | - +-------------------------------+-------------------------------+ - | Masking-key (continued) | Payload Data | - +-------------------------------- - - - - - - - - - - - - - - - + - : Payload Data continued : - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + - | Payload Data continued | - +-----------------------------------------------------------------------+ - @param buffer - @param exbuffers - @return + * 0 1 2 3 + * 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 + * +-+-+-+-+-------+-+-------------+-------------------------------+ + * |F|R|R|R| opcode|M| Payload len | Extended payload length | + * |I|S|S|S| (4) |A| (7) | (16/64) | + * |N|V|V|V| |S| | (if payload len==126/127) | + * | |1|2|3| |K| | | + * +-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - + + * | Extended payload length continued, if payload len == 127 | + * + - - - - - - - - - - - - - - - +-------------------------------+ + * | |Masking-key, if MASK set to 1 | + * +-------------------------------+-------------------------------+ + * | Masking-key (continued) | Payload Data | + * +-------------------------------- - - - - - - - - - - - - - - - + + * : Payload Data continued : + * + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + + * | Payload Data continued | + * +-----------------------------------------------------------------------+ + * + * @param buffer + * @param exbuffers + * @return */ public WebSocketPacket decode(final ByteBuffer buffer, ByteBuffer... exbuffers) { final boolean debug = this.debugable; diff --git a/src/org/redkale/net/http/WebSocketServlet.java b/src/org/redkale/net/http/WebSocketServlet.java index 01664bb35..b0c81cf29 100644 --- a/src/org/redkale/net/http/WebSocketServlet.java +++ b/src/org/redkale/net/http/WebSocketServlet.java @@ -63,7 +63,7 @@ public abstract class WebSocketServlet extends HttpServlet { @Override public void init(Context context, AnyValue conf) { InetSocketAddress addr = context.getServerAddress(); - this.engine = new WebSocketEngine(addr.getHostString() + ":" + addr.getPort() + "-" + name(), logger); + this.engine = new WebSocketEngine(addr.getHostString() + ":" + addr.getPort() + "-" + name(), this.node, logger); this.node.putWebSocketEngine(engine); this.node.init(conf); this.engine.init(conf); @@ -97,15 +97,14 @@ public abstract class WebSocketServlet extends HttpServlet { return; } final WebSocket webSocket = this.createWebSocket(); - webSocket.engine = engine; - webSocket.node = node; + webSocket._engine = engine; Serializable sessionid = webSocket.onOpen(request); if (sessionid == null) { if (debug) logger.finest("WebSocket connect abort, Not found sessionid. request=" + request); response.finish(true); return; } - webSocket.sessionid = sessionid; + webSocket._sessionid = sessionid; request.setKeepAlive(true); byte[] bytes = (key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").getBytes(); synchronized (digest) { @@ -127,7 +126,7 @@ public abstract class WebSocketServlet extends HttpServlet { response.finish(true); return; } - webSocket.groupid = groupid; + webSocket._groupid = groupid; engine.add(webSocket); context.submit(new WebSocketRunner(context, webSocket, response.removeChannel(), wsbinary)); response.finish(true);