WebSocket支持permessage-deflate单向功能

src/org/redkale/net/http/WebSocket.java

@@ -14,6 +14,7 @@ import java.util.concurrent.*;
 import java.util.function.*;
 import java.util.logging.*;
 import java.util.stream.Stream;
+import java.util.zip.*;
 import org.redkale.convert.Convert;
 import org.redkale.net.AsyncConnection;
 import org.redkale.util.Comment;
@@ -103,7 +104,11 @@ public abstract class WebSocket<G extends Serializable, T> {
 
     java.lang.reflect.Type _messageTextType; //不可能为空
 
-    private long createtime = System.currentTimeMillis();
+    Deflater deflater; //压缩
+
+    Inflater inflater; //解压
+
+    long createtime = System.currentTimeMillis();
 
     private long pingtime;
 
@@ -894,6 +899,8 @@ public abstract class WebSocket<G extends Serializable, T> {
      * 显式地关闭WebSocket
      */
     public final void close() {
+        if (this.deflater != null) this.deflater.end();
+        if (this.inflater != null) this.inflater.end();
         if (this._runner != null) {
             CompletableFuture<Void> future = this._runner.closeRunner(CLOSECODE_SERVERCLOSE, "user close");
             if (future != null) future.join();

src/org/redkale/net/http/WebSocketEngine.java

@@ -244,7 +244,7 @@ public class WebSocketEngine {
                     if (bufferSupplier == null) {
                         bufferSupplier = websocket.getBufferSupplier();
                         bufferConsumer = websocket.getBufferConsumer();
-                        packet.setSendBuffers(packet.encode(bufferSupplier, bufferConsumer, cryptor));
+                        packet.setSendBuffers(packet.encodePacket(bufferSupplier, bufferConsumer, cryptor));
                     }
                     future = future == null ? websocket.sendPacket(packet) : future.thenCombine(websocket.sendPacket(packet), (a, b) -> a | (Integer) b);
                 }
@@ -255,7 +255,7 @@ public class WebSocketEngine {
                         if (bufferSupplier == null) {
                             bufferSupplier = websocket.getBufferSupplier();
                             bufferConsumer = websocket.getBufferConsumer();
-                            packet.setSendBuffers(packet.encode(bufferSupplier, bufferConsumer, cryptor));
+                            packet.setSendBuffers(packet.encodePacket(bufferSupplier, bufferConsumer, cryptor));
                         }
                         future = future == null ? websocket.sendPacket(packet) : future.thenCombine(websocket.sendPacket(packet), (a, b) -> a | (Integer) b);
                     }
@@ -321,7 +321,7 @@ public class WebSocketEngine {
                     if (bufferSupplier == null) {
                         bufferSupplier = websocket.getBufferSupplier();
                         bufferConsumer = websocket.getBufferConsumer();
-                        packet.setSendBuffers(packet.encode(bufferSupplier, bufferConsumer, cryptor));
+                        packet.setSendBuffers(packet.encodePacket(bufferSupplier, bufferConsumer, cryptor));
                     }
                     future = future == null ? websocket.sendPacket(packet) : future.thenCombine(websocket.sendPacket(packet), (a, b) -> a | (Integer) b);
                 }
@@ -333,7 +333,7 @@ public class WebSocketEngine {
                         if (bufferSupplier == null) {
                             bufferSupplier = websocket.getBufferSupplier();
                             bufferConsumer = websocket.getBufferConsumer();
-                            packet.setSendBuffers(packet.encode(bufferSupplier, bufferConsumer, cryptor));
+                            packet.setSendBuffers(packet.encodePacket(bufferSupplier, bufferConsumer, cryptor));
                         }
                         future = future == null ? websocket.sendPacket(packet) : future.thenCombine(websocket.sendPacket(packet), (a, b) -> a | (Integer) b);
                     }

src/org/redkale/net/http/WebSocketPacket.java

@@ -9,9 +9,10 @@ import org.redkale.util.Utility;
 import java.io.*;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
-import java.util.AbstractMap;
+import java.util.*;
 import java.util.function.*;
 import java.util.logging.*;
+import java.util.zip.*;
 import org.redkale.convert.*;
 import org.redkale.net.Cryptor;
 import org.redkale.util.*;
@@ -29,6 +30,8 @@ public final class WebSocketPacket {
 
     static final WebSocketPacket NONE = new WebSocketPacket();
 
+    private static final byte[] EOM_BYTES = new byte[]{0, 0, -1, -1};
+
     public static final WebSocketPacket DEFAULT_PING_PACKET = new WebSocketPacket(FrameType.PING, new byte[0]);
 
     public static enum MessageType {
@@ -82,6 +85,8 @@ public final class WebSocketPacket {
     //---------------接收------------------------
     MessageType receiveType;
 
+    boolean receiveCompress;
+
     int receiveCount;
 
     int receiveLength;
@@ -223,7 +228,7 @@ public final class WebSocketPacket {
      *
      * @return ByteBuffer[]
      */
-    ByteBuffer[] encode(final Supplier<ByteBuffer> supplier, final Consumer<ByteBuffer> consumer, final Cryptor cryptor) {
+    ByteBuffer[] encodePacket(final Supplier<ByteBuffer> supplier, final Consumer<ByteBuffer> consumer, final Cryptor cryptor) {
         final byte opcode = (byte) (this.type.getValue() | 0x80);
         if (this.sendConvert != null) {
             Supplier<ByteBuffer> newsupplier = new Supplier<ByteBuffer>() {
@@ -391,7 +396,7 @@ public final class WebSocketPacket {
      *
      * @return 返回NONE表示Buffer内容不够； 返回this表示解析完成或部分解析完成；返回null表示解析异常；
      */
-    WebSocketPacket decode(final Logger logger, final WebSocketRunner runner, final WebSocket webSocket, final int wsmaxbody,
+    WebSocketPacket decodePacket(final Logger logger, final WebSocketRunner runner, final WebSocket webSocket, final int wsmaxbody,
         final AbstractMap.SimpleEntry<String, byte[]> halfBytes, final ByteBuffer buffer) {
         //开始
         final boolean debug = false; //调试开关
@@ -407,6 +412,17 @@ public final class WebSocketPacket {
         this.last = (opcode & 0b1000_0000) != 0;
         this.type = FrameType.valueOf(opcode & 0xF);
 
+        //0x00 表示一个后续帧 
+        //0x01 表示一个文本帧 
+        //0x02 表示一个二进制帧 
+        //0x03-07 为以后的非控制帧保留
+        //0x8 表示一个连接关闭
+        //0x9 表示一个ping
+        //0xA 表示一个pong
+        //0x0B-0F 为以后的控制帧保留
+        final boolean control = (opcode & 0b0000_1000) != 0; //是否控制帧
+        this.receiveCompress = !control && webSocket.deflater != null && (opcode & 0b0100_0000) != 0; //rsv1 为 1
+
         if (type == FrameType.CLOSE) {
             if (debug) logger.log(Level.FINEST, " receive close command from websocket client");
         }
@@ -418,15 +434,6 @@ public final class WebSocketPacket {
             if (debug) logger.log(Level.FINE, "rsv1 rsv2 rsv3 must be 0, but not (" + opcode + ")");
             return null; //rsv1 rsv2 rsv3 must be 0     
         }
-        //0x00 表示一个后续帧 
-        //0x01 表示一个文本帧 
-        //0x02 表示一个二进制帧 
-        //0x03-07 为以后的非控制帧保留
-        //0x8 表示一个连接关闭
-        //0x9 表示一个ping
-        //0xA 表示一个pong
-        //0x0B-0F 为以后的控制帧保留
-        final boolean control = (opcode & 0b0000_1000) != 0; //是否控制帧
         final byte crcode = buffer.get();  //第二个字节
 
         byte lengthCode = crcode;
@@ -506,14 +513,15 @@ public final class WebSocketPacket {
         if (selfType == FrameType.TEXT) {
             Convert textConvert = webSocket.getTextConvert();
             if (textConvert == null || (!runner.mergemsg && (series || !this.last))) {
-                this.receiveMessage = new String(this.getReceiveBytes(buffers), StandardCharsets.UTF_8);
+                this.receiveMessage = new String(this.getReceiveBytes(webSocket, buffers), StandardCharsets.UTF_8);
                 this.receiveType = MessageType.STRING;
             } else {
                 if (this.last || !runner.mergemsg) {
-                    if (runner.currSeriesMergeMessage == null) {
+                    if (runner.currSeriesMergeMessage == null && !this.receiveCompress) {
                         this.receiveMessage = textConvert.convertFrom(webSocket._messageTextType, this.receiveMasker, buffers);
                     } else {
-                        runner.currSeriesMergeMessage.write(this.getReceiveBytes(buffers));
+                        if (runner.currSeriesMergeMessage == null) runner.currSeriesMergeMessage = new ByteArray();
+                        runner.currSeriesMergeMessage.write(this.getReceiveBytes(webSocket, buffers));
                         try {
                             this.receiveMessage = textConvert.convertFrom(webSocket._messageTextType, runner.currSeriesMergeMessage.getBytes());
                         } finally {
@@ -522,7 +530,7 @@ public final class WebSocketPacket {
                     }
                 } else {
                     if (runner.currSeriesMergeMessage == null) runner.currSeriesMergeMessage = new ByteArray();
-                    runner.currSeriesMergeMessage.write(this.getReceiveBytes(buffers));
+                    runner.currSeriesMergeMessage.write(this.getReceiveBytes(webSocket, buffers));
                     this.receiveMessage = MESSAGE_NIL;
                 }
                 this.receiveCount = this.receiveLength;
@@ -531,14 +539,15 @@ public final class WebSocketPacket {
         } else if (selfType == FrameType.BINARY) {
             Convert binaryConvert = webSocket.getBinaryConvert();
             if (binaryConvert == null || (!runner.mergemsg && (series || !this.last))) {
-                this.receiveMessage = this.getReceiveBytes(buffers);
+                this.receiveMessage = this.getReceiveBytes(webSocket, buffers);
                 this.receiveType = MessageType.BYTES;
             } else {
                 if (this.last || !runner.mergemsg) {
-                    if (runner.currSeriesMergeMessage == null) {
+                    if (runner.currSeriesMergeMessage == null && !this.receiveCompress) {
                         this.receiveMessage = binaryConvert.convertFrom(webSocket._messageTextType, this.receiveMasker, buffers);
                     } else {
-                        runner.currSeriesMergeMessage.write(this.getReceiveBytes(buffers));
+                        if (runner.currSeriesMergeMessage == null) runner.currSeriesMergeMessage = new ByteArray();
+                        runner.currSeriesMergeMessage.write(this.getReceiveBytes(webSocket, buffers));
                         try {
                             this.receiveMessage = binaryConvert.convertFrom(webSocket._messageTextType, runner.currSeriesMergeMessage.getBytes());
                         } finally {
@@ -547,20 +556,20 @@ public final class WebSocketPacket {
                     }
                 } else {
                     if (runner.currSeriesMergeMessage == null) runner.currSeriesMergeMessage = new ByteArray();
-                    runner.currSeriesMergeMessage.write(this.getReceiveBytes(buffers));
+                    runner.currSeriesMergeMessage.write(this.getReceiveBytes(webSocket, buffers));
                     this.receiveMessage = MESSAGE_NIL;
                 }
                 this.receiveCount = this.receiveLength;
                 this.receiveType = MessageType.OBJECT;
             }
         } else if (selfType == FrameType.PING) {
-            this.receiveMessage = this.getReceiveBytes(buffers);
+            this.receiveMessage = this.getReceiveBytes(webSocket, buffers);
             this.receiveType = MessageType.BYTES;
         } else if (selfType == FrameType.PONG) {
-            this.receiveMessage = this.getReceiveBytes(buffers);
+            this.receiveMessage = this.getReceiveBytes(webSocket, buffers);
             this.receiveType = MessageType.BYTES;
         } else if (selfType == FrameType.CLOSE) {
-            this.receiveMessage = this.getReceiveBytes(buffers);
+            this.receiveMessage = this.getReceiveBytes(webSocket, buffers);
             this.receiveType = MessageType.BYTES;
         }
     }
@@ -569,7 +578,7 @@ public final class WebSocketPacket {
         return this.receiveLength <= this.receiveCount;
     }
 
-    byte[] getReceiveBytes(ByteBuffer... buffers) {
+    byte[] getReceiveBytes(WebSocket webSocket, ByteBuffer... buffers) {
         final int length = this.receiveLength;
         if (length == 0) return new byte[0];
         byte[] bs = new byte[length];
@@ -587,6 +596,23 @@ public final class WebSocketPacket {
                 bs[i] = mask.unmask(bs[i]);
             }
         }
+
+        if (this.receiveCompress) {
+            Inflater inflater = new Inflater(true);
+            ByteArrayOutputStream baos = new ByteArrayOutputStream(bs.length);
+            inflater.setInput(Utility.append(bs, EOM_BYTES));
+            byte[] buff = new byte[1024];
+            try {
+                while (!inflater.finished()) {
+                    int count = inflater.inflate(buff);
+                    if (count == 0) break;
+                    baos.write(buff, 0, count);
+                }
+                return baos.toByteArray();
+            } catch (Exception ex) {
+                return bs;
+            }
+        }
         return bs;
     }
 

src/org/redkale/net/http/WebSocketRunner.java

@@ -108,7 +108,7 @@ class WebSocketRunner implements Runnable {
                             if (onePacket != null) packets.add(onePacket);
                             try {
                                 while (true) {
-                                    WebSocketPacket packet = new WebSocketPacket().decode(context.getLogger(), self, webSocket, wsmaxbody, halfBytes, readBuffer);
+                                    WebSocketPacket packet = new WebSocketPacket().decodePacket(context.getLogger(), self, webSocket, wsmaxbody, halfBytes, readBuffer);
                                     if (packet == WebSocketPacket.NONE) break; //解析完毕但是buffer有多余字节
                                     if (packet != null && !packet.isReceiveFinished()) {
                                         unfinishPacket = packet;
@@ -230,7 +230,7 @@ class WebSocketRunner implements Runnable {
         //System.out.println("推送消息");        
         final CompletableFuture<Integer> futureResult = new CompletableFuture<>();
         try {
-            ByteBuffer[] buffers = packet.sendBuffers != null ? packet.duplicateSendBuffers() : packet.encode(webSocket._channel.getBufferSupplier(), webSocket._channel.getBufferConsumer(), webSocket._engine.cryptor);
+            ByteBuffer[] buffers = packet.sendBuffers != null ? packet.duplicateSendBuffers() : packet.encodePacket(webSocket._channel.getBufferSupplier(), webSocket._channel.getBufferConsumer(), webSocket._engine.cryptor);
             //if (debug) context.getLogger().log(Level.FINEST, "wsrunner.sending websocket message:  " + packet);
             CompletionHandler<Integer, ByteBuffer[]> handler = new CompletionHandler<Integer, ByteBuffer[]>() {
 

src/org/redkale/net/http/WebSocketServlet.java

@@ -15,6 +15,7 @@ import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.*;
 import java.util.logging.*;
+import java.util.zip.*;
 import javax.annotation.*;
 import org.redkale.convert.Convert;
 import org.redkale.net.Cryptor;
@@ -210,6 +211,10 @@ public abstract class WebSocketServlet extends HttpServlet implements Resourcabl
         webSocket._remoteAddress = request.getRemoteAddress();
         webSocket._remoteAddr = request.getRemoteAddr();
         webSocket._sncpAddress = this.node.localSncpAddress;
+        if (request.getHeader("Sec-WebSocket-Extensions", "").contains("permessage-deflate")) {
+            webSocket.deflater = new Deflater(Deflater.DEFAULT_COMPRESSION, true);
+            webSocket.inflater = new Inflater(true);
+        }
         initRestWebSocket(webSocket);
         CompletableFuture<String> sessionFuture = webSocket.onOpen(request);
         if (sessionFuture == null) {
@@ -233,6 +238,8 @@ public abstract class WebSocketServlet extends HttpServlet implements Resourcabl
             response.setHeader("Connection", "Upgrade");
             response.addHeader("Upgrade", "websocket");
             response.addHeader("Sec-WebSocket-Accept", Base64.getEncoder().encodeToString(bytes));
+            if (webSocket.deflater != null) response.addHeader("Sec-WebSocket-Extensions", "permessage-deflate");
+
             response.sendBody((ByteBuffer) null, null, new CompletionHandler<Integer, Void>() {
 
                 WebSocketRunner temprunner = null;

src/org/redkale/util/Redkale.java

@@ -17,7 +17,7 @@ public final class Redkale {
     }
 
     public static String getDotedVersion() {
-        return "2.0.0";  
+        return "2.1.0";  
     }
 
     public static int getMajorVersion() {
@@ -25,6 +25,6 @@ public final class Redkale {
     }
 
     public static int getMinorVersion() { 
-        return 0;
+        return 1;
     }
 }