Add user possibility to skip the evaluation of a certain websocket ex… (#8910)

Motivation:

Add user possibility to skip the evaluation of certain web socket extension,
for example we can skip compression extension for messages that already compressed or very small and etc.

Modification:

This pull request is related with #5669

Result:

User can set to WebSocketClientExtensionHandshaker or WebSocketServerExtensionHandshaker a filter to skip the evaluation of certain extension.

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/BinaryWebSocketFrame.java

@@ -19,7 +19,7 @@ import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 
 /**
- * Web Socket frame containing binary data
+ * Web Socket frame containing binary data.
  */
 public class BinaryWebSocketFrame extends WebSocketFrame {
 

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/CloseWebSocketFrame.java

@@ -21,7 +21,7 @@ import io.netty.util.CharsetUtil;
 import io.netty.util.internal.StringUtil;
 
 /**
- * Web Socket Frame for closing the connection
+ * Web Socket Frame for closing the connection.
  */
 public class CloseWebSocketFrame extends WebSocketFrame {
 
@@ -51,7 +51,7 @@ public class CloseWebSocketFrame extends WebSocketFrame {
      * @param finalFragment
      *            flag indicating if this frame is the final fragment
      * @param rsv
-     *            reserved bits used for protocol extensions
+     *            reserved bits used for protocol extensions.
      */
     public CloseWebSocketFrame(boolean finalFragment, int rsv) {
         this(finalFragment, rsv, Unpooled.buffer(0));

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/ContinuationWebSocketFrame.java

@@ -43,7 +43,7 @@ public class ContinuationWebSocketFrame extends WebSocketFrame {
     }
 
     /**
-     * Creates a new continuation frame with the specified binary data
+     * Creates a new continuation frame with the specified binary data.
      *
      * @param finalFragment
      *            flag indicating if this frame is the final fragment
@@ -71,17 +71,17 @@ public class ContinuationWebSocketFrame extends WebSocketFrame {
     }
 
     /**
-     * Returns the text data in this frame
+     * Returns the text data in this frame.
      */
     public String text() {
         return content().toString(CharsetUtil.UTF_8);
     }
 
     /**
-     * Sets the string for this frame
+     * Sets the string for this frame.
      *
      * @param text
-     *            text to store
+     *            text to store.
      */
     private static ByteBuf fromText(String text) {
         if (text == null || text.isEmpty()) {

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/PingWebSocketFrame.java

@@ -19,7 +19,7 @@ import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 
 /**
- * Web Socket frame containing binary data
+ * Web Socket frame containing binary data.
  */
 public class PingWebSocketFrame extends WebSocketFrame {
 
@@ -41,7 +41,7 @@ public class PingWebSocketFrame extends WebSocketFrame {
     }
 
     /**
-     * Creates a new ping frame with the specified binary data
+     * Creates a new ping frame with the specified binary data.
      *
      * @param finalFragment
      *            flag indicating if this frame is the final fragment

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/PongWebSocketFrame.java

@@ -19,7 +19,7 @@ import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 
 /**
- * Web Socket frame containing binary data
+ * Web Socket frame containing binary data.
  */
 public class PongWebSocketFrame extends WebSocketFrame {
 

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/TextWebSocketFrame.java

@@ -20,7 +20,7 @@ import io.netty.buffer.Unpooled;
 import io.netty.util.CharsetUtil;
 
 /**
- * Web Socket text frame
+ * Web Socket text frame.
  */
 public class TextWebSocketFrame extends WebSocketFrame {
 
@@ -35,7 +35,7 @@ public class TextWebSocketFrame extends WebSocketFrame {
      * Creates a new text frame with the specified text string. The final fragment flag is set to true.
      *
      * @param text
-     *            String to put in the frame
+     *            String to put in the frame.
      */
     public TextWebSocketFrame(String text) {
         super(fromText(text));
@@ -59,7 +59,7 @@ public class TextWebSocketFrame extends WebSocketFrame {
      * @param rsv
      *            reserved bits used for protocol extensions
      * @param text
-     *            String to put in the frame
+     *            String to put in the frame.
      */
     public TextWebSocketFrame(boolean finalFragment, int rsv, String text) {
         super(finalFragment, rsv, fromText(text));
@@ -74,7 +74,7 @@ public class TextWebSocketFrame extends WebSocketFrame {
     }
 
     /**
-     * Creates a new text frame with the specified binary data. The final fragment flag is set to true.
+     * Creates a new text frame with the specified binary data and the final fragment flag.
      *
      * @param finalFragment
      *            flag indicating if this frame is the final fragment
@@ -88,7 +88,7 @@ public class TextWebSocketFrame extends WebSocketFrame {
     }
 
     /**
-     * Returns the text data in this frame
+     * Returns the text data in this frame.
      */
     public String text() {
         return content().toString(CharsetUtil.UTF_8);

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketFrame.java

@@ -20,7 +20,7 @@ import io.netty.buffer.DefaultByteBufHolder;
 import io.netty.util.internal.StringUtil;
 
 /**
- * Base class for web socket frames
+ * Base class for web socket frames.
  */
 public abstract class WebSocketFrame extends DefaultByteBufHolder {
 

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/extensions/WebSocketExtensionFilter.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2019 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.handler.codec.http.websocketx.extensions;
+
+import io.netty.handler.codec.http.websocketx.WebSocketFrame;
+
+/**
+ * Filter that is responsible to skip the evaluation of a certain extension
+ * according to standard.
+ */
+public interface WebSocketExtensionFilter {
+
+    /**
+     * A {@link WebSocketExtensionFilter} that never skip the evaluation of an
+     * any given extensions {@link WebSocketExtension}.
+     */
+    WebSocketExtensionFilter NEVER_SKIP = new WebSocketExtensionFilter() {
+        @Override
+        public boolean mustSkip(WebSocketFrame frame) {
+            return false;
+        }
+    };
+
+    /**
+     * A {@link WebSocketExtensionFilter} that always skip the evaluation of an
+     * any given extensions {@link WebSocketExtension}.
+     */
+    WebSocketExtensionFilter ALWAYS_SKIP = new WebSocketExtensionFilter() {
+        @Override
+        public boolean mustSkip(WebSocketFrame frame) {
+            return true;
+        }
+    };
+
+    /**
+     * Returns {@code true} if the evaluation of the extension must skipped
+     * for the given frame otherwise {@code false}.
+     */
+    boolean mustSkip(WebSocketFrame frame);
+
+}

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/extensions/WebSocketExtensionFilterProvider.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2019 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.handler.codec.http.websocketx.extensions;
+
+/**
+ * Extension filter provider that is responsible to provide filters for a certain {@link WebSocketExtension} extension.
+ */
+public interface WebSocketExtensionFilterProvider {
+
+    WebSocketExtensionFilterProvider DEFAULT = new WebSocketExtensionFilterProvider() {
+        @Override
+        public WebSocketExtensionFilter encoderFilter() {
+            return WebSocketExtensionFilter.NEVER_SKIP;
+        }
+
+        @Override
+        public WebSocketExtensionFilter decoderFilter() {
+            return WebSocketExtensionFilter.NEVER_SKIP;
+        }
+    };
+
+    /**
+     * Returns the extension filter for {@link WebSocketExtensionEncoder} encoder.
+     */
+    WebSocketExtensionFilter encoderFilter();
+
+    /**
+     * Returns the extension filter for {@link WebSocketExtensionDecoder} decoder.
+     */
+    WebSocketExtensionFilter decoderFilter();
+
+}

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/extensions/compression/DeflateDecoder.java

@@ -28,9 +28,12 @@ import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionDecoder;
+import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionFilter;
 
 import java.util.List;
 
+import static io.netty.util.internal.ObjectUtil.*;
+
 /**
  * Deflate implementation of a payload decompressor for
  * <tt>io.netty.handler.codec.http.websocketx.WebSocketFrame</tt>.
@@ -40,15 +43,26 @@ abstract class DeflateDecoder extends WebSocketExtensionDecoder {
     static final byte[] FRAME_TAIL = new byte[] {0x00, 0x00, (byte) 0xff, (byte) 0xff};
 
     private final boolean noContext;
+    private final WebSocketExtensionFilter extensionDecoderFilter;
 
     private EmbeddedChannel decoder;
 
     /**
      * Constructor
+     *
      * @param noContext true to disable context takeover.
+     * @param extensionDecoderFilter extension decoder filter.
      */
-    DeflateDecoder(boolean noContext) {
+    DeflateDecoder(boolean noContext, WebSocketExtensionFilter extensionDecoderFilter) {
         this.noContext = noContext;
+        this.extensionDecoderFilter = checkNotNull(extensionDecoderFilter, "extensionDecoderFilter");
+    }
+
+    /**
+     * Returns the extension decoder filter.
+     */
+    protected WebSocketExtensionFilter extensionDecoderFilter() {
+        return extensionDecoderFilter;
     }
 
     protected abstract boolean appendFrameTail(WebSocketFrame msg);

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/extensions/compression/DeflateEncoder.java

@@ -15,7 +15,6 @@
  */
 package io.netty.handler.codec.http.websocketx.extensions.compression;
 
-import static io.netty.handler.codec.http.websocketx.extensions.compression.PerMessageDeflateDecoder.*;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.CompositeByteBuf;
 import io.netty.channel.ChannelHandlerContext;
@@ -28,9 +27,13 @@ import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionEncoder;
+import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionFilter;
 
 import java.util.List;
 
+import static io.netty.handler.codec.http.websocketx.extensions.compression.PerMessageDeflateDecoder.*;
+import static io.netty.util.internal.ObjectUtil.*;
+
 /**
  * Deflate implementation of a payload compressor for
  * <tt>io.netty.handler.codec.http.websocketx.WebSocketFrame</tt>.
@@ -40,6 +43,7 @@ abstract class DeflateEncoder extends WebSocketExtensionEncoder {
     private final int compressionLevel;
     private final int windowSize;
     private final boolean noContext;
+    private final WebSocketExtensionFilter extensionEncoderFilter;
 
     private EmbeddedChannel encoder;
 
@@ -48,11 +52,21 @@ abstract class DeflateEncoder extends WebSocketExtensionEncoder {
      * @param compressionLevel compression level of the compressor.
      * @param windowSize maximum size of the window compressor buffer.
      * @param noContext true to disable context takeover.
+     * @param extensionEncoderFilter extension encoder filter.
      */
-    DeflateEncoder(int compressionLevel, int windowSize, boolean noContext) {
+    DeflateEncoder(int compressionLevel, int windowSize, boolean noContext,
+                   WebSocketExtensionFilter extensionEncoderFilter) {
         this.compressionLevel = compressionLevel;
         this.windowSize = windowSize;
         this.noContext = noContext;
+        this.extensionEncoderFilter = checkNotNull(extensionEncoderFilter, "extensionEncoderFilter");
+    }
+
+    /**
+     * Returns the extension encoder filter.
+     */
+    protected WebSocketExtensionFilter extensionEncoderFilter() {
+        return extensionEncoderFilter;
     }
 
     /**

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/extensions/compression/DeflateFrameClientExtensionHandshaker.java

@@ -15,16 +15,18 @@
  */
 package io.netty.handler.codec.http.websocketx.extensions.compression;
 
-import static io.netty.handler.codec.http.websocketx.extensions.compression.
-        DeflateFrameServerExtensionHandshaker.*;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketClientExtension;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketClientExtensionHandshaker;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionData;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionDecoder;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionEncoder;
+import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionFilterProvider;
 
 import java.util.Collections;
 
+import static io.netty.handler.codec.http.websocketx.extensions.compression.DeflateFrameServerExtensionHandshaker.*;
+import static io.netty.util.internal.ObjectUtil.*;
+
 /**
  * <a href="https://tools.ietf.org/id/draft-tyoshino-hybi-websocket-perframe-deflate-06.txt">perframe-deflate</a>
  * handshake implementation.
@@ -33,6 +35,7 @@ public final class DeflateFrameClientExtensionHandshaker implements WebSocketCli
 
     private final int compressionLevel;
     private final boolean useWebkitExtensionName;
+    private final WebSocketExtensionFilterProvider extensionFilterProvider;
 
     /**
      * Constructor with default configuration.
@@ -48,12 +51,26 @@ public final class DeflateFrameClientExtensionHandshaker implements WebSocketCli
      *            Compression level between 0 and 9 (default is 6).
      */
     public DeflateFrameClientExtensionHandshaker(int compressionLevel, boolean useWebkitExtensionName) {
+        this(compressionLevel, useWebkitExtensionName, WebSocketExtensionFilterProvider.DEFAULT);
+    }
+
+    /**
+     * Constructor with custom configuration.
+     *
+     * @param compressionLevel
+     *            Compression level between 0 and 9 (default is 6).
+     * @param extensionFilterProvider
+     *            provides client extension filters for per frame deflate encoder and decoder.
+     */
+    public DeflateFrameClientExtensionHandshaker(int compressionLevel, boolean useWebkitExtensionName,
+            WebSocketExtensionFilterProvider extensionFilterProvider) {
         if (compressionLevel < 0 || compressionLevel > 9) {
             throw new IllegalArgumentException(
                     "compressionLevel: " + compressionLevel + " (expected: 0-9)");
         }
         this.compressionLevel = compressionLevel;
         this.useWebkitExtensionName = useWebkitExtensionName;
+        this.extensionFilterProvider = checkNotNull(extensionFilterProvider, "extensionFilterProvider");
     }
 
     @Override
@@ -71,7 +88,7 @@ public final class DeflateFrameClientExtensionHandshaker implements WebSocketCli
         }
 
         if (extensionData.parameters().isEmpty()) {
-            return new DeflateFrameClientExtension(compressionLevel);
+            return new DeflateFrameClientExtension(compressionLevel, extensionFilterProvider);
         } else {
             return null;
         }
@@ -80,9 +97,11 @@ public final class DeflateFrameClientExtensionHandshaker implements WebSocketCli
     private static class DeflateFrameClientExtension implements WebSocketClientExtension {
 
         private final int compressionLevel;
+        private final WebSocketExtensionFilterProvider extensionFilterProvider;
 
-        DeflateFrameClientExtension(int compressionLevel) {
+        DeflateFrameClientExtension(int compressionLevel, WebSocketExtensionFilterProvider extensionFilterProvider) {
             this.compressionLevel = compressionLevel;
+            this.extensionFilterProvider = extensionFilterProvider;
         }
 
         @Override
@@ -92,12 +111,13 @@ public final class DeflateFrameClientExtensionHandshaker implements WebSocketCli
 
         @Override
         public WebSocketExtensionEncoder newExtensionEncoder() {
-            return new PerFrameDeflateEncoder(compressionLevel, 15, false);
+            return new PerFrameDeflateEncoder(compressionLevel, 15, false,
+                                              extensionFilterProvider.encoderFilter());
         }
 
         @Override
         public WebSocketExtensionDecoder newExtensionDecoder() {
-            return new PerFrameDeflateDecoder(false);
+            return new PerFrameDeflateDecoder(false, extensionFilterProvider.decoderFilter());
         }
     }
 

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/extensions/compression/DeflateFrameServerExtensionHandshaker.java

@@ -18,11 +18,14 @@ package io.netty.handler.codec.http.websocketx.extensions.compression;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionData;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionDecoder;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionEncoder;
+import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionFilterProvider;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketServerExtension;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketServerExtensionHandshaker;
 
 import java.util.Collections;
 
+import static io.netty.util.internal.ObjectUtil.*;
+
 /**
  * <a href="https://tools.ietf.org/id/draft-tyoshino-hybi-websocket-perframe-deflate-06.txt">perframe-deflate</a>
  * handshake implementation.
@@ -33,6 +36,7 @@ public final class DeflateFrameServerExtensionHandshaker implements WebSocketSer
     static final String DEFLATE_FRAME_EXTENSION = "deflate-frame";
 
     private final int compressionLevel;
+    private final WebSocketExtensionFilterProvider extensionFilterProvider;
 
     /**
      * Constructor with default configuration.
@@ -48,11 +52,25 @@ public final class DeflateFrameServerExtensionHandshaker implements WebSocketSer
      *            Compression level between 0 and 9 (default is 6).
      */
     public DeflateFrameServerExtensionHandshaker(int compressionLevel) {
+        this(compressionLevel, WebSocketExtensionFilterProvider.DEFAULT);
+    }
+
+    /**
+     * Constructor with custom configuration.
+     *
+     * @param compressionLevel
+     *            Compression level between 0 and 9 (default is 6).
+     * @param extensionFilterProvider
+     *            provides server extension filters for per frame deflate encoder and decoder.
+     */
+    public DeflateFrameServerExtensionHandshaker(int compressionLevel,
+            WebSocketExtensionFilterProvider extensionFilterProvider) {
         if (compressionLevel < 0 || compressionLevel > 9) {
             throw new IllegalArgumentException(
                     "compressionLevel: " + compressionLevel + " (expected: 0-9)");
         }
         this.compressionLevel = compressionLevel;
+        this.extensionFilterProvider = checkNotNull(extensionFilterProvider, "extensionFilterProvider");
     }
 
     @Override
@@ -63,7 +81,7 @@ public final class DeflateFrameServerExtensionHandshaker implements WebSocketSer
         }
 
         if (extensionData.parameters().isEmpty()) {
-            return new DeflateFrameServerExtension(compressionLevel, extensionData.name());
+            return new DeflateFrameServerExtension(compressionLevel, extensionData.name(), extensionFilterProvider);
         } else {
             return null;
         }
@@ -73,10 +91,13 @@ public final class DeflateFrameServerExtensionHandshaker implements WebSocketSer
 
         private final String extensionName;
         private final int compressionLevel;
+        private final WebSocketExtensionFilterProvider extensionFilterProvider;
 
-        DeflateFrameServerExtension(int compressionLevel, String extensionName) {
+        DeflateFrameServerExtension(int compressionLevel, String extensionName,
+                WebSocketExtensionFilterProvider extensionFilterProvider) {
             this.extensionName = extensionName;
             this.compressionLevel = compressionLevel;
+            this.extensionFilterProvider = extensionFilterProvider;
         }
 
         @Override
@@ -86,12 +107,13 @@ public final class DeflateFrameServerExtensionHandshaker implements WebSocketSer
 
         @Override
         public WebSocketExtensionEncoder newExtensionEncoder() {
-            return new PerFrameDeflateEncoder(compressionLevel, 15, false);
+            return new PerFrameDeflateEncoder(compressionLevel, 15, false,
+                                              extensionFilterProvider.encoderFilter());
         }
 
         @Override
         public WebSocketExtensionDecoder newExtensionDecoder() {
-            return new PerFrameDeflateDecoder(false);
+            return new PerFrameDeflateDecoder(false, extensionFilterProvider.decoderFilter());
         }
 
         @Override

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/extensions/compression/PerFrameDeflateDecoder.java

@@ -20,6 +20,7 @@ import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtension;
+import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionFilter;
 
 /**
  * Per-frame implementation of deflate decompressor.
@@ -28,18 +29,37 @@ class PerFrameDeflateDecoder extends DeflateDecoder {
 
     /**
      * Constructor
+     *
      * @param noContext true to disable context takeover.
      */
     PerFrameDeflateDecoder(boolean noContext) {
-        super(noContext);
+        super(noContext, WebSocketExtensionFilter.NEVER_SKIP);
+    }
+
+    /**
+     * Constructor
+     *
+     * @param noContext true to disable context takeover.
+     * @param extensionDecoderFilter extension decoder filter for per frame deflate decoder.
+     */
+    PerFrameDeflateDecoder(boolean noContext, WebSocketExtensionFilter extensionDecoderFilter) {
+        super(noContext, extensionDecoderFilter);
     }
 
     @Override
     public boolean acceptInboundMessage(Object msg) throws Exception {
-        return (msg instanceof TextWebSocketFrame ||
+        if (!super.acceptInboundMessage(msg)) {
-                msg instanceof BinaryWebSocketFrame ||
+            return false;
+        }
+
+        WebSocketFrame wsFrame = (WebSocketFrame) msg;
+        if (extensionDecoderFilter().mustSkip(wsFrame)) {
+            return false;
+        }
+
+        return (msg instanceof TextWebSocketFrame || msg instanceof BinaryWebSocketFrame ||
                 msg instanceof ContinuationWebSocketFrame) &&
-                    (((WebSocketFrame) msg).rsv() & WebSocketExtension.RSV1) > 0;
+               (wsFrame.rsv() & WebSocketExtension.RSV1) > 0;
     }
 
     @Override
@@ -51,4 +71,5 @@ class PerFrameDeflateDecoder extends DeflateDecoder {
     protected boolean appendFrameTail(WebSocketFrame msg) {
         return true;
     }
+
 }

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/extensions/compression/PerFrameDeflateEncoder.java

@@ -20,6 +20,7 @@ import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtension;
+import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionFilter;
 
 /**
  * Per-frame implementation of deflate compressor.
@@ -28,21 +29,43 @@ class PerFrameDeflateEncoder extends DeflateEncoder {
 
     /**
      * Constructor
+     *
      * @param compressionLevel compression level of the compressor.
-     * @param windowSize maximum size of the window compressor buffer.
+     * @param windowSize       maximum size of the window compressor buffer.
-     * @param noContext true to disable context takeover.
+     * @param noContext        true to disable context takeover.
      */
     PerFrameDeflateEncoder(int compressionLevel, int windowSize, boolean noContext) {
-        super(compressionLevel, windowSize, noContext);
+        super(compressionLevel, windowSize, noContext, WebSocketExtensionFilter.NEVER_SKIP);
+    }
+
+    /**
+     * Constructor
+     *
+     * @param compressionLevel compression level of the compressor.
+     * @param windowSize       maximum size of the window compressor buffer.
+     * @param noContext        true to disable context takeover.
+     * @param extensionEncoderFilter extension encoder filter for per frame deflate encoder.
+     */
+    PerFrameDeflateEncoder(int compressionLevel, int windowSize, boolean noContext,
+                           WebSocketExtensionFilter extensionEncoderFilter) {
+        super(compressionLevel, windowSize, noContext, extensionEncoderFilter);
     }
 
     @Override
     public boolean acceptOutboundMessage(Object msg) throws Exception {
-        return (msg instanceof TextWebSocketFrame ||
+        if (!super.acceptOutboundMessage(msg)) {
-                msg instanceof BinaryWebSocketFrame ||
+            return false;
+        }
+
+        WebSocketFrame wsFrame = (WebSocketFrame) msg;
+        if (extensionEncoderFilter().mustSkip(wsFrame)) {
+            return false;
+        }
+
+        return (msg instanceof TextWebSocketFrame || msg instanceof BinaryWebSocketFrame ||
                 msg instanceof ContinuationWebSocketFrame) &&
-                    ((WebSocketFrame) msg).content().readableBytes() > 0 &&
+               wsFrame.content().readableBytes() > 0 &&
-                    (((WebSocketFrame) msg).rsv() & WebSocketExtension.RSV1) == 0;
+               (wsFrame.rsv() & WebSocketExtension.RSV1) == 0;
     }
 
     @Override

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/extensions/compression/PerMessageDeflateClientExtensionHandshaker.java

@@ -15,20 +15,21 @@
  */
 package io.netty.handler.codec.http.websocketx.extensions.compression;
 
-import static io.netty.handler.codec.http.websocketx.extensions.compression.
-        PerMessageDeflateServerExtensionHandshaker.*;
-
 import io.netty.handler.codec.compression.ZlibCodecFactory;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketClientExtension;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketClientExtensionHandshaker;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionData;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionDecoder;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionEncoder;
+import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionFilterProvider;
 
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map.Entry;
 
+import static io.netty.handler.codec.http.websocketx.extensions.compression.PerMessageDeflateServerExtensionHandshaker.*;
+import static io.netty.util.internal.ObjectUtil.*;
+
 /**
  * <a href="http://tools.ietf.org/html/draft-ietf-hybi-permessage-compression-18">permessage-deflate</a>
  * handshake implementation.
@@ -40,6 +41,7 @@ public final class PerMessageDeflateClientExtensionHandshaker implements WebSock
     private final int requestedServerWindowSize;
     private final boolean allowClientNoContext;
     private final boolean requestedServerNoContext;
+    private final WebSocketExtensionFilterProvider extensionFilterProvider;
 
     /**
      * Constructor with default configuration.
@@ -68,6 +70,34 @@ public final class PerMessageDeflateClientExtensionHandshaker implements WebSock
     public PerMessageDeflateClientExtensionHandshaker(int compressionLevel,
             boolean allowClientWindowSize, int requestedServerWindowSize,
             boolean allowClientNoContext, boolean requestedServerNoContext) {
+        this(compressionLevel, allowClientWindowSize, requestedServerWindowSize,
+             allowClientNoContext, requestedServerNoContext, WebSocketExtensionFilterProvider.DEFAULT);
+    }
+
+    /**
+     * Constructor with custom configuration.
+     *
+     * @param compressionLevel
+     *            Compression level between 0 and 9 (default is 6).
+     * @param allowClientWindowSize
+     *            allows WebSocket server to customize the client inflater window size
+     *            (default is false).
+     * @param requestedServerWindowSize
+     *            indicates the requested sever window size to use if server inflater is customizable.
+     * @param allowClientNoContext
+     *            allows WebSocket server to activate client_no_context_takeover
+     *            (default is false).
+     * @param requestedServerNoContext
+     *            indicates if client needs to activate server_no_context_takeover
+     *            if server is compatible with (default is false).
+     * @param extensionFilterProvider
+     *            provides client extension filters for per message deflate encoder and decoder.
+     */
+    public PerMessageDeflateClientExtensionHandshaker(int compressionLevel,
+            boolean allowClientWindowSize, int requestedServerWindowSize,
+            boolean allowClientNoContext, boolean requestedServerNoContext,
+            WebSocketExtensionFilterProvider extensionFilterProvider) {
+
         if (requestedServerWindowSize > MAX_WINDOW_SIZE || requestedServerWindowSize < MIN_WINDOW_SIZE) {
             throw new IllegalArgumentException(
                     "requestedServerWindowSize: " + requestedServerWindowSize + " (expected: 8-15)");
@@ -81,6 +111,7 @@ public final class PerMessageDeflateClientExtensionHandshaker implements WebSock
         this.requestedServerWindowSize = requestedServerWindowSize;
         this.allowClientNoContext = allowClientNoContext;
         this.requestedServerNoContext = requestedServerNoContext;
+        this.extensionFilterProvider = checkNotNull(extensionFilterProvider, "extensionFilterProvider");
     }
 
     @Override
@@ -158,7 +189,7 @@ public final class PerMessageDeflateClientExtensionHandshaker implements WebSock
 
         if (succeed) {
             return new PermessageDeflateExtension(serverNoContext, serverWindowSize,
-                    clientNoContext, clientWindowSize);
+                    clientNoContext, clientWindowSize, extensionFilterProvider);
         } else {
             return null;
         }
@@ -170,6 +201,7 @@ public final class PerMessageDeflateClientExtensionHandshaker implements WebSock
         private final int serverWindowSize;
         private final boolean clientNoContext;
         private final int clientWindowSize;
+        private final WebSocketExtensionFilterProvider extensionFilterProvider;
 
         @Override
         public int rsv() {
@@ -177,21 +209,24 @@ public final class PerMessageDeflateClientExtensionHandshaker implements WebSock
         }
 
         PermessageDeflateExtension(boolean serverNoContext, int serverWindowSize,
-                boolean clientNoContext, int clientWindowSize) {
+                boolean clientNoContext, int clientWindowSize,
+                WebSocketExtensionFilterProvider extensionFilterProvider) {
             this.serverNoContext = serverNoContext;
             this.serverWindowSize = serverWindowSize;
             this.clientNoContext = clientNoContext;
             this.clientWindowSize = clientWindowSize;
+            this.extensionFilterProvider = extensionFilterProvider;
         }
 
         @Override
         public WebSocketExtensionEncoder newExtensionEncoder() {
-            return new PerMessageDeflateEncoder(compressionLevel, clientWindowSize, clientNoContext);
+            return new PerMessageDeflateEncoder(compressionLevel, clientWindowSize, clientNoContext,
+                                                extensionFilterProvider.encoderFilter());
         }
 
         @Override
         public WebSocketExtensionDecoder newExtensionDecoder() {
-            return new PerMessageDeflateDecoder(serverNoContext);
+            return new PerMessageDeflateDecoder(serverNoContext, extensionFilterProvider.decoderFilter());
         }
     }
 

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/extensions/compression/PerMessageDeflateDecoder.java

@@ -21,6 +21,7 @@ import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtension;
+import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionFilter;
 
 import java.util.List;
 
@@ -33,23 +34,45 @@ class PerMessageDeflateDecoder extends DeflateDecoder {
 
     /**
      * Constructor
+     *
      * @param noContext true to disable context takeover.
      */
     PerMessageDeflateDecoder(boolean noContext) {
-        super(noContext);
+        super(noContext, WebSocketExtensionFilter.NEVER_SKIP);
+    }
+
+    /**
+     * Constructor
+     *
+     * @param noContext true to disable context takeover.
+     * @param extensionDecoderFilter extension decoder for per message deflate decoder.
+     */
+    PerMessageDeflateDecoder(boolean noContext, WebSocketExtensionFilter extensionDecoderFilter) {
+        super(noContext, extensionDecoderFilter);
     }
 
     @Override
     public boolean acceptInboundMessage(Object msg) throws Exception {
-        return ((msg instanceof TextWebSocketFrame ||
+        if (!super.acceptInboundMessage(msg)) {
-                 msg instanceof BinaryWebSocketFrame) &&
+            return false;
-                    (((WebSocketFrame) msg).rsv() & WebSocketExtension.RSV1) > 0) ||
+        }
-                (msg instanceof ContinuationWebSocketFrame && compressing);
+
+        WebSocketFrame wsFrame = (WebSocketFrame) msg;
+        if (extensionDecoderFilter().mustSkip(wsFrame)) {
+            if (compressing) {
+                throw new IllegalStateException("Cannot skip per message deflate decoder, compression in progress");
+            }
+            return false;
+        }
+
+        return ((wsFrame instanceof TextWebSocketFrame || wsFrame instanceof BinaryWebSocketFrame) &&
+                (wsFrame.rsv() & WebSocketExtension.RSV1) > 0) ||
+               (wsFrame instanceof ContinuationWebSocketFrame && compressing);
     }
 
     @Override
     protected int newRsv(WebSocketFrame msg) {
-        return (msg.rsv() & WebSocketExtension.RSV1) > 0 ?
+        return (msg.rsv() & WebSocketExtension.RSV1) > 0?
                 msg.rsv() ^ WebSocketExtension.RSV1 : msg.rsv();
     }
 
@@ -60,7 +83,7 @@ class PerMessageDeflateDecoder extends DeflateDecoder {
 
     @Override
     protected void decode(ChannelHandlerContext ctx, WebSocketFrame msg,
-        List<Object> out) throws Exception {
+                          List<Object> out) throws Exception {
         super.decode(ctx, msg, out);
 
         if (msg.isFinalFragment()) {
@@ -69,4 +92,5 @@ class PerMessageDeflateDecoder extends DeflateDecoder {
             compressing = true;
         }
     }
+
 }

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/extensions/compression/PerMessageDeflateEncoder.java

@@ -21,6 +21,7 @@ import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtension;
+import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionFilter;
 
 import java.util.List;
 
@@ -33,25 +34,50 @@ class PerMessageDeflateEncoder extends DeflateEncoder {
 
     /**
      * Constructor
+     *
      * @param compressionLevel compression level of the compressor.
      * @param windowSize maximum size of the window compressor buffer.
      * @param noContext true to disable context takeover.
      */
     PerMessageDeflateEncoder(int compressionLevel, int windowSize, boolean noContext) {
-        super(compressionLevel, windowSize, noContext);
+        super(compressionLevel, windowSize, noContext, WebSocketExtensionFilter.NEVER_SKIP);
+    }
+
+    /**
+     * Constructor
+     *
+     * @param compressionLevel compression level of the compressor.
+     * @param windowSize maximum size of the window compressor buffer.
+     * @param noContext true to disable context takeover.
+     * @param extensionEncoderFilter extension filter for per message deflate encoder.
+     */
+    PerMessageDeflateEncoder(int compressionLevel, int windowSize, boolean noContext,
+                             WebSocketExtensionFilter extensionEncoderFilter) {
+        super(compressionLevel, windowSize, noContext, extensionEncoderFilter);
     }
 
     @Override
     public boolean acceptOutboundMessage(Object msg) throws Exception {
-        return ((msg instanceof TextWebSocketFrame ||
+        if (!super.acceptOutboundMessage(msg)) {
-                msg instanceof BinaryWebSocketFrame) &&
+            return false;
-                   (((WebSocketFrame) msg).rsv() & WebSocketExtension.RSV1) == 0) ||
+        }
-               (msg instanceof ContinuationWebSocketFrame && compressing);
+
+        WebSocketFrame wsFrame = (WebSocketFrame) msg;
+        if (extensionEncoderFilter().mustSkip(wsFrame)) {
+            if (compressing) {
+                throw new IllegalStateException("Cannot skip per message deflate encoder, compression in progress");
+            }
+            return false;
+        }
+
+        return ((wsFrame instanceof TextWebSocketFrame || wsFrame instanceof BinaryWebSocketFrame) &&
+                (wsFrame.rsv() & WebSocketExtension.RSV1) == 0) ||
+               (wsFrame instanceof ContinuationWebSocketFrame && compressing);
     }
 
     @Override
     protected int rsv(WebSocketFrame msg) {
-        return msg instanceof TextWebSocketFrame || msg instanceof BinaryWebSocketFrame ?
+        return msg instanceof TextWebSocketFrame || msg instanceof BinaryWebSocketFrame?
                 msg.rsv() | WebSocketExtension.RSV1 : msg.rsv();
     }
 
@@ -62,7 +88,7 @@ class PerMessageDeflateEncoder extends DeflateEncoder {
 
     @Override
     protected void encode(ChannelHandlerContext ctx, WebSocketFrame msg,
-            List<Object> out) throws Exception {
+                          List<Object> out) throws Exception {
         super.encode(ctx, msg, out);
 
         if (msg.isFinalFragment()) {

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/extensions/compression/PerMessageDeflateServerExtensionHandshaker.java

@@ -19,6 +19,7 @@ import io.netty.handler.codec.compression.ZlibCodecFactory;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionData;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionDecoder;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionEncoder;
+import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionFilterProvider;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketServerExtension;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketServerExtensionHandshaker;
 
@@ -26,6 +27,8 @@ import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map.Entry;
 
+import static io.netty.util.internal.ObjectUtil.*;
+
 /**
  * <a href="http://tools.ietf.org/html/draft-ietf-hybi-permessage-compression-18">permessage-deflate</a>
  * handshake implementation.
@@ -46,6 +49,7 @@ public final class PerMessageDeflateServerExtensionHandshaker implements WebSock
     private final int preferredClientWindowSize;
     private final boolean allowServerNoContext;
     private final boolean preferredClientNoContext;
+    private final WebSocketExtensionFilterProvider extensionFilterProvider;
 
     /**
      * Constructor with default configuration.
@@ -71,9 +75,36 @@ public final class PerMessageDeflateServerExtensionHandshaker implements WebSock
      *            indicates if server prefers to activate client_no_context_takeover
      *            if client is compatible with (default is false).
      */
-    public PerMessageDeflateServerExtensionHandshaker(int compressionLevel,
+    public PerMessageDeflateServerExtensionHandshaker(int compressionLevel, boolean allowServerWindowSize,
-            boolean allowServerWindowSize, int preferredClientWindowSize,
+            int preferredClientWindowSize,
             boolean allowServerNoContext, boolean preferredClientNoContext) {
+        this(compressionLevel, allowServerWindowSize, preferredClientWindowSize, allowServerNoContext,
+             preferredClientNoContext, WebSocketExtensionFilterProvider.DEFAULT);
+    }
+
+    /**
+     * Constructor with custom configuration.
+     *
+     * @param compressionLevel
+     *            Compression level between 0 and 9 (default is 6).
+     * @param allowServerWindowSize
+     *            allows WebSocket client to customize the server inflater window size
+     *            (default is false).
+     * @param preferredClientWindowSize
+     *            indicates the preferred client window size to use if client inflater is customizable.
+     * @param allowServerNoContext
+     *            allows WebSocket client to activate server_no_context_takeover
+     *            (default is false).
+     * @param preferredClientNoContext
+     *            indicates if server prefers to activate client_no_context_takeover
+     *            if client is compatible with (default is false).
+     * @param extensionFilterProvider
+     *            provides server extension filters for per message deflate encoder and decoder.
+     */
+    public PerMessageDeflateServerExtensionHandshaker(int compressionLevel, boolean allowServerWindowSize,
+            int preferredClientWindowSize,
+            boolean allowServerNoContext, boolean preferredClientNoContext,
+            WebSocketExtensionFilterProvider extensionFilterProvider) {
         if (preferredClientWindowSize > MAX_WINDOW_SIZE || preferredClientWindowSize < MIN_WINDOW_SIZE) {
             throw new IllegalArgumentException(
                     "preferredServerWindowSize: " + preferredClientWindowSize + " (expected: 8-15)");
@@ -87,6 +118,7 @@ public final class PerMessageDeflateServerExtensionHandshaker implements WebSock
         this.preferredClientWindowSize = preferredClientWindowSize;
         this.allowServerNoContext = allowServerNoContext;
         this.preferredClientNoContext = preferredClientNoContext;
+        this.extensionFilterProvider = checkNotNull(extensionFilterProvider, "extensionFilterProvider");
     }
 
     @Override
@@ -137,7 +169,7 @@ public final class PerMessageDeflateServerExtensionHandshaker implements WebSock
 
         if (deflateEnabled) {
             return new PermessageDeflateExtension(compressionLevel, serverNoContext,
-                    serverWindowSize, clientNoContext, clientWindowSize);
+                    serverWindowSize, clientNoContext, clientWindowSize, extensionFilterProvider);
         } else {
             return null;
         }
@@ -150,14 +182,17 @@ public final class PerMessageDeflateServerExtensionHandshaker implements WebSock
         private final int serverWindowSize;
         private final boolean clientNoContext;
         private final int clientWindowSize;
+        private final WebSocketExtensionFilterProvider extensionFilterProvider;
 
         PermessageDeflateExtension(int compressionLevel, boolean serverNoContext,
-                int serverWindowSize, boolean clientNoContext, int clientWindowSize) {
+                int serverWindowSize, boolean clientNoContext, int clientWindowSize,
+                WebSocketExtensionFilterProvider extensionFilterProvider) {
             this.compressionLevel = compressionLevel;
             this.serverNoContext = serverNoContext;
             this.serverWindowSize = serverWindowSize;
             this.clientNoContext = clientNoContext;
             this.clientWindowSize = clientWindowSize;
+            this.extensionFilterProvider = extensionFilterProvider;
         }
 
         @Override
@@ -167,12 +202,13 @@ public final class PerMessageDeflateServerExtensionHandshaker implements WebSock
 
         @Override
         public WebSocketExtensionEncoder newExtensionEncoder() {
-            return new PerMessageDeflateEncoder(compressionLevel, serverWindowSize, serverNoContext);
+            return new PerMessageDeflateEncoder(compressionLevel, serverWindowSize, serverNoContext,
+                                                extensionFilterProvider.encoderFilter());
         }
 
         @Override
         public WebSocketExtensionDecoder newExtensionDecoder() {
-            return new PerMessageDeflateDecoder(clientNoContext);
+            return new PerMessageDeflateDecoder(clientNoContext, extensionFilterProvider.decoderFilter());
         }
 
         @Override

codec-http/src/test/java/io/netty/handler/codec/http/websocketx/extensions/WebSocketExtensionFilterProviderTest.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2019 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.handler.codec.http.websocketx.extensions;
+
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+public class WebSocketExtensionFilterProviderTest {
+
+    @Test
+    public void testDefaultExtensionFilterProvider() {
+        WebSocketExtensionFilterProvider defaultProvider = WebSocketExtensionFilterProvider.DEFAULT;
+        assertNotNull(defaultProvider);
+
+        assertEquals(WebSocketExtensionFilter.NEVER_SKIP, defaultProvider.decoderFilter());
+        assertEquals(WebSocketExtensionFilter.NEVER_SKIP, defaultProvider.encoderFilter());
+    }
+}

codec-http/src/test/java/io/netty/handler/codec/http/websocketx/extensions/WebSocketExtensionFilterTest.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2019 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.handler.codec.http.websocketx.extensions;
+
+import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+public class WebSocketExtensionFilterTest {
+
+    @Test
+    public void testNeverSkip() {
+        WebSocketExtensionFilter neverSkip = WebSocketExtensionFilter.NEVER_SKIP;
+
+        BinaryWebSocketFrame binaryFrame = new BinaryWebSocketFrame();
+        assertFalse(neverSkip.mustSkip(binaryFrame));
+        assertTrue(binaryFrame.release());
+
+        TextWebSocketFrame textFrame = new TextWebSocketFrame();
+        assertFalse(neverSkip.mustSkip(textFrame));
+        assertTrue(textFrame.release());
+
+        PingWebSocketFrame pingFrame = new PingWebSocketFrame();
+        assertFalse(neverSkip.mustSkip(pingFrame));
+        assertTrue(pingFrame.release());
+
+        PongWebSocketFrame pongFrame = new PongWebSocketFrame();
+        assertFalse(neverSkip.mustSkip(pongFrame));
+        assertTrue(pongFrame.release());
+
+        CloseWebSocketFrame closeFrame = new CloseWebSocketFrame();
+        assertFalse(neverSkip.mustSkip(closeFrame));
+        assertTrue(closeFrame.release());
+
+        ContinuationWebSocketFrame continuationFrame = new ContinuationWebSocketFrame();
+        assertFalse(neverSkip.mustSkip(continuationFrame));
+        assertTrue(continuationFrame.release());
+    }
+
+    @Test
+    public void testAlwaysSkip() {
+        WebSocketExtensionFilter neverSkip = WebSocketExtensionFilter.ALWAYS_SKIP;
+
+        BinaryWebSocketFrame binaryFrame = new BinaryWebSocketFrame();
+        assertTrue(neverSkip.mustSkip(binaryFrame));
+        assertTrue(binaryFrame.release());
+
+        TextWebSocketFrame textFrame = new TextWebSocketFrame();
+        assertTrue(neverSkip.mustSkip(textFrame));
+        assertTrue(textFrame.release());
+
+        PingWebSocketFrame pingFrame = new PingWebSocketFrame();
+        assertTrue(neverSkip.mustSkip(pingFrame));
+        assertTrue(pingFrame.release());
+
+        PongWebSocketFrame pongFrame = new PongWebSocketFrame();
+        assertTrue(neverSkip.mustSkip(pongFrame));
+        assertTrue(pongFrame.release());
+
+        CloseWebSocketFrame closeFrame = new CloseWebSocketFrame();
+        assertTrue(neverSkip.mustSkip(closeFrame));
+        assertTrue(closeFrame.release());
+
+        ContinuationWebSocketFrame continuationFrame = new ContinuationWebSocketFrame();
+        assertTrue(neverSkip.mustSkip(continuationFrame));
+        assertTrue(continuationFrame.release());
+    }
+}

codec-http/src/test/java/io/netty/handler/codec/http/websocketx/extensions/compression/PerFrameDeflateDecoderTest.java

@@ -15,20 +15,20 @@
  */
 package io.netty.handler.codec.http.websocketx.extensions.compression;
 
-import static io.netty.handler.codec.http.websocketx.extensions.WebSocketExtension.RSV1;
-import static io.netty.handler.codec.http.websocketx.extensions.WebSocketExtension.RSV3;
-import static org.junit.Assert.*;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.handler.codec.compression.ZlibCodecFactory;
 import io.netty.handler.codec.compression.ZlibWrapper;
 import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtension;
+import org.junit.Test;
 
-import java.util.Arrays;
 import java.util.Random;
 
-import org.junit.Test;
+import static io.netty.handler.codec.http.websocketx.extensions.WebSocketExtension.*;
+import static io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionFilter.*;
+import static org.junit.Assert.*;
 
 public class PerFrameDeflateDecoderTest {
 
@@ -44,7 +44,7 @@ public class PerFrameDeflateDecoderTest {
         byte[] payload = new byte[300];
         random.nextBytes(payload);
 
-        encoderChannel.writeOutbound(Unpooled.wrappedBuffer(payload));
+        assertTrue(encoderChannel.writeOutbound(Unpooled.wrappedBuffer(payload)));
         ByteBuf compressedPayload = encoderChannel.readOutbound();
 
         BinaryWebSocketFrame compressedFrame = new BinaryWebSocketFrame(true,
@@ -52,19 +52,18 @@ public class PerFrameDeflateDecoderTest {
                 compressedPayload.slice(0, compressedPayload.readableBytes() - 4));
 
         // execute
-        decoderChannel.writeInbound(compressedFrame);
+        assertTrue(decoderChannel.writeInbound(compressedFrame));
         BinaryWebSocketFrame uncompressedFrame = decoderChannel.readInbound();
 
         // test
         assertNotNull(uncompressedFrame);
         assertNotNull(uncompressedFrame.content());
-        assertTrue(uncompressedFrame instanceof BinaryWebSocketFrame);
         assertEquals(RSV3, uncompressedFrame.rsv());
         assertEquals(300, uncompressedFrame.content().readableBytes());
 
         byte[] finalPayload = new byte[300];
         uncompressedFrame.content().readBytes(finalPayload);
-        assertTrue(Arrays.equals(finalPayload, payload));
+        assertArrayEquals(finalPayload, payload);
         uncompressedFrame.release();
     }
 
@@ -80,19 +79,18 @@ public class PerFrameDeflateDecoderTest {
                 RSV3, Unpooled.wrappedBuffer(payload));
 
         // execute
-        decoderChannel.writeInbound(frame);
+        assertTrue(decoderChannel.writeInbound(frame));
         BinaryWebSocketFrame newFrame = decoderChannel.readInbound();
 
         // test
         assertNotNull(newFrame);
         assertNotNull(newFrame.content());
-        assertTrue(newFrame instanceof BinaryWebSocketFrame);
         assertEquals(RSV3, newFrame.rsv());
         assertEquals(300, newFrame.content().readableBytes());
 
         byte[] finalPayload = new byte[300];
         newFrame.content().readBytes(finalPayload);
-        assertTrue(Arrays.equals(finalPayload, payload));
+        assertArrayEquals(finalPayload, payload);
         newFrame.release();
     }
 
@@ -103,21 +101,51 @@ public class PerFrameDeflateDecoderTest {
                 ZlibCodecFactory.newZlibEncoder(ZlibWrapper.NONE, 9, 15, 8));
         EmbeddedChannel decoderChannel = new EmbeddedChannel(new PerFrameDeflateDecoder(false));
 
-        encoderChannel.writeOutbound(Unpooled.EMPTY_BUFFER);
+        assertTrue(encoderChannel.writeOutbound(Unpooled.EMPTY_BUFFER));
         ByteBuf compressedPayload = encoderChannel.readOutbound();
         BinaryWebSocketFrame compressedFrame =
                 new BinaryWebSocketFrame(true, RSV1 | RSV3, compressedPayload);
 
         // execute
-        decoderChannel.writeInbound(compressedFrame);
+        assertTrue(decoderChannel.writeInbound(compressedFrame));
         BinaryWebSocketFrame uncompressedFrame = decoderChannel.readInbound();
 
         // test
         assertNotNull(uncompressedFrame);
         assertNotNull(uncompressedFrame.content());
-        assertTrue(uncompressedFrame instanceof BinaryWebSocketFrame);
         assertEquals(RSV3, uncompressedFrame.rsv());
         assertEquals(0, uncompressedFrame.content().readableBytes());
         uncompressedFrame.release();
     }
+
+    @Test
+    public void testDecompressionSkip() {
+        EmbeddedChannel encoderChannel = new EmbeddedChannel(
+                ZlibCodecFactory.newZlibEncoder(ZlibWrapper.NONE, 9, 15, 8));
+        EmbeddedChannel decoderChannel = new EmbeddedChannel(new PerFrameDeflateDecoder(false, ALWAYS_SKIP));
+
+        byte[] payload = new byte[300];
+        random.nextBytes(payload);
+
+        assertTrue(encoderChannel.writeOutbound(Unpooled.wrappedBuffer(payload)));
+        ByteBuf compressedPayload = encoderChannel.readOutbound();
+
+        BinaryWebSocketFrame compressedBinaryFrame = new BinaryWebSocketFrame(
+                true, WebSocketExtension.RSV1 | WebSocketExtension.RSV3, compressedPayload);
+
+        assertTrue(decoderChannel.writeInbound(compressedBinaryFrame));
+
+        BinaryWebSocketFrame inboundBinaryFrame = decoderChannel.readInbound();
+
+        assertNotNull(inboundBinaryFrame);
+        assertNotNull(inboundBinaryFrame.content());
+        assertEquals(compressedPayload, inboundBinaryFrame.content());
+        assertEquals(5, inboundBinaryFrame.rsv());
+
+        assertTrue(inboundBinaryFrame.release());
+
+        assertTrue(encoderChannel.finishAndReleaseAll());
+        assertFalse(decoderChannel.finish());
+    }
+
 }

codec-http/src/test/java/io/netty/handler/codec/http/websocketx/extensions/compression/PerFrameDeflateEncoderTest.java

@@ -15,8 +15,8 @@
  */
 package io.netty.handler.codec.http.websocketx.extensions.compression;
 
-import static org.junit.Assert.*;
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.handler.codec.compression.ZlibCodecFactory;
@@ -24,11 +24,12 @@ import io.netty.handler.codec.compression.ZlibWrapper;
 import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtension;
+import org.junit.Test;
 
-import java.util.Arrays;
 import java.util.Random;
 
-import org.junit.Test;
+import static io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionFilter.*;
+import static org.junit.Assert.*;
 
 public class PerFrameDeflateEncoderTest {
 
@@ -47,23 +48,22 @@ public class PerFrameDeflateEncoderTest {
                 WebSocketExtension.RSV3, Unpooled.wrappedBuffer(payload));
 
         // execute
-        encoderChannel.writeOutbound(frame);
+        assertTrue(encoderChannel.writeOutbound(frame));
         BinaryWebSocketFrame compressedFrame = encoderChannel.readOutbound();
 
         // test
         assertNotNull(compressedFrame);
         assertNotNull(compressedFrame.content());
-        assertTrue(compressedFrame instanceof BinaryWebSocketFrame);
         assertEquals(WebSocketExtension.RSV1 | WebSocketExtension.RSV3, compressedFrame.rsv());
 
-        decoderChannel.writeInbound(compressedFrame.content());
+        assertTrue(decoderChannel.writeInbound(compressedFrame.content()));
-        decoderChannel.writeInbound(DeflateDecoder.FRAME_TAIL);
+        assertTrue(decoderChannel.writeInbound(Unpooled.wrappedBuffer(DeflateDecoder.FRAME_TAIL)));
         ByteBuf uncompressedPayload = decoderChannel.readInbound();
         assertEquals(300, uncompressedPayload.readableBytes());
 
         byte[] finalPayload = new byte[300];
         uncompressedPayload.readBytes(finalPayload);
-        assertTrue(Arrays.equals(finalPayload, payload));
+        assertArrayEquals(finalPayload, payload);
         uncompressedPayload.release();
     }
 
@@ -79,19 +79,18 @@ public class PerFrameDeflateEncoderTest {
                 WebSocketExtension.RSV3 | WebSocketExtension.RSV1, Unpooled.wrappedBuffer(payload));
 
         // execute
-        encoderChannel.writeOutbound(frame);
+        assertTrue(encoderChannel.writeOutbound(frame));
         BinaryWebSocketFrame newFrame = encoderChannel.readOutbound();
 
         // test
         assertNotNull(newFrame);
         assertNotNull(newFrame.content());
-        assertTrue(newFrame instanceof BinaryWebSocketFrame);
         assertEquals(WebSocketExtension.RSV3 | WebSocketExtension.RSV1, newFrame.rsv());
         assertEquals(300, newFrame.content().readableBytes());
 
         byte[] finalPayload = new byte[300];
         newFrame.content().readBytes(finalPayload);
-        assertTrue(Arrays.equals(finalPayload, payload));
+        assertArrayEquals(finalPayload, payload);
         newFrame.release();
     }
 
@@ -117,9 +116,9 @@ public class PerFrameDeflateEncoderTest {
                 WebSocketExtension.RSV3, Unpooled.wrappedBuffer(payload3));
 
         // execute
-        encoderChannel.writeOutbound(frame1);
+        assertTrue(encoderChannel.writeOutbound(frame1));
-        encoderChannel.writeOutbound(frame2);
+        assertTrue(encoderChannel.writeOutbound(frame2));
-        encoderChannel.writeOutbound(frame3);
+        assertTrue(encoderChannel.writeOutbound(frame3));
         BinaryWebSocketFrame compressedFrame1 = encoderChannel.readOutbound();
         ContinuationWebSocketFrame compressedFrame2 = encoderChannel.readOutbound();
         ContinuationWebSocketFrame compressedFrame3 = encoderChannel.readOutbound();
@@ -135,28 +134,52 @@ public class PerFrameDeflateEncoderTest {
         assertFalse(compressedFrame2.isFinalFragment());
         assertTrue(compressedFrame3.isFinalFragment());
 
-        decoderChannel.writeInbound(compressedFrame1.content());
+        assertTrue(decoderChannel.writeInbound(compressedFrame1.content()));
-        decoderChannel.writeInbound(Unpooled.wrappedBuffer(DeflateDecoder.FRAME_TAIL));
+        assertTrue(decoderChannel.writeInbound(Unpooled.wrappedBuffer(DeflateDecoder.FRAME_TAIL)));
         ByteBuf uncompressedPayload1 = decoderChannel.readInbound();
         byte[] finalPayload1 = new byte[100];
         uncompressedPayload1.readBytes(finalPayload1);
-        assertTrue(Arrays.equals(finalPayload1, payload1));
+        assertArrayEquals(finalPayload1, payload1);
         uncompressedPayload1.release();
 
-        decoderChannel.writeInbound(compressedFrame2.content());
+        assertTrue(decoderChannel.writeInbound(compressedFrame2.content()));
-        decoderChannel.writeInbound(Unpooled.wrappedBuffer(DeflateDecoder.FRAME_TAIL));
+        assertTrue(decoderChannel.writeInbound(Unpooled.wrappedBuffer(DeflateDecoder.FRAME_TAIL)));
         ByteBuf uncompressedPayload2 = decoderChannel.readInbound();
         byte[] finalPayload2 = new byte[100];
         uncompressedPayload2.readBytes(finalPayload2);
-        assertTrue(Arrays.equals(finalPayload2, payload2));
+        assertArrayEquals(finalPayload2, payload2);
         uncompressedPayload2.release();
 
-        decoderChannel.writeInbound(compressedFrame3.content());
+        assertTrue(decoderChannel.writeInbound(compressedFrame3.content()));
-        decoderChannel.writeInbound(Unpooled.wrappedBuffer(DeflateDecoder.FRAME_TAIL));
+        assertTrue(decoderChannel.writeInbound(Unpooled.wrappedBuffer(DeflateDecoder.FRAME_TAIL)));
         ByteBuf uncompressedPayload3 = decoderChannel.readInbound();
         byte[] finalPayload3 = new byte[100];
         uncompressedPayload3.readBytes(finalPayload3);
-        assertTrue(Arrays.equals(finalPayload3, payload3));
+        assertArrayEquals(finalPayload3, payload3);
         uncompressedPayload3.release();
     }
+
+    @Test
+    public void testCompressionSkip() {
+        EmbeddedChannel encoderChannel = new EmbeddedChannel(
+                new PerFrameDeflateEncoder(9, 15, false, ALWAYS_SKIP));
+        byte[] payload = new byte[300];
+        random.nextBytes(payload);
+        BinaryWebSocketFrame binaryFrame = new BinaryWebSocketFrame(true,
+                                                                    0, Unpooled.wrappedBuffer(payload));
+
+        // execute
+        assertTrue(encoderChannel.writeOutbound(binaryFrame.copy()));
+        BinaryWebSocketFrame outboundFrame = encoderChannel.readOutbound();
+
+        // test
+        assertNotNull(outboundFrame);
+        assertNotNull(outboundFrame.content());
+        assertArrayEquals(payload, ByteBufUtil.getBytes(outboundFrame.content()));
+        assertEquals(0, outboundFrame.rsv());
+        assertTrue(outboundFrame.release());
+
+        assertFalse(encoderChannel.finish());
+    }
+
 }

codec-http/src/test/java/io/netty/handler/codec/http/websocketx/extensions/compression/PerMessageDeflateDecoderTest.java

@@ -15,20 +15,26 @@
  */
 package io.netty.handler.codec.http.websocketx.extensions.compression;
 
-import static org.junit.Assert.*;
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.DecoderException;
 import io.netty.handler.codec.compression.ZlibCodecFactory;
 import io.netty.handler.codec.compression.ZlibWrapper;
 import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.WebSocketFrame;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtension;
+import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionFilter;
+import org.junit.Test;
 
-import java.util.Arrays;
 import java.util.Random;
 
-import org.junit.Test;
+import static io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionFilter.*;
+import static io.netty.util.CharsetUtil.*;
+import static org.junit.Assert.*;
 
 public class PerMessageDeflateDecoderTest {
 
@@ -44,7 +50,7 @@ public class PerMessageDeflateDecoderTest {
         byte[] payload = new byte[300];
         random.nextBytes(payload);
 
-        encoderChannel.writeOutbound(Unpooled.wrappedBuffer(payload));
+        assertTrue(encoderChannel.writeOutbound(Unpooled.wrappedBuffer(payload)));
         ByteBuf compressedPayload = encoderChannel.readOutbound();
 
         BinaryWebSocketFrame compressedFrame = new BinaryWebSocketFrame(true,
@@ -52,19 +58,18 @@ public class PerMessageDeflateDecoderTest {
                 compressedPayload.slice(0, compressedPayload.readableBytes() - 4));
 
         // execute
-        decoderChannel.writeInbound(compressedFrame);
+        assertTrue(decoderChannel.writeInbound(compressedFrame));
         BinaryWebSocketFrame uncompressedFrame = decoderChannel.readInbound();
 
         // test
         assertNotNull(uncompressedFrame);
         assertNotNull(uncompressedFrame.content());
-        assertTrue(uncompressedFrame instanceof BinaryWebSocketFrame);
         assertEquals(WebSocketExtension.RSV3, uncompressedFrame.rsv());
         assertEquals(300, uncompressedFrame.content().readableBytes());
 
         byte[] finalPayload = new byte[300];
         uncompressedFrame.content().readBytes(finalPayload);
-        assertTrue(Arrays.equals(finalPayload, payload));
+        assertArrayEquals(finalPayload, payload);
         uncompressedFrame.release();
     }
 
@@ -80,24 +85,23 @@ public class PerMessageDeflateDecoderTest {
                 WebSocketExtension.RSV3, Unpooled.wrappedBuffer(payload));
 
         // execute
-        decoderChannel.writeInbound(frame);
+        assertTrue(decoderChannel.writeInbound(frame));
         BinaryWebSocketFrame newFrame = decoderChannel.readInbound();
 
         // test
         assertNotNull(newFrame);
         assertNotNull(newFrame.content());
-        assertTrue(newFrame instanceof BinaryWebSocketFrame);
         assertEquals(WebSocketExtension.RSV3, newFrame.rsv());
         assertEquals(300, newFrame.content().readableBytes());
 
         byte[] finalPayload = new byte[300];
         newFrame.content().readBytes(finalPayload);
-        assertTrue(Arrays.equals(finalPayload, payload));
+        assertArrayEquals(finalPayload, payload);
         newFrame.release();
     }
 
     @Test
-    public void testFramementedFrame() {
+    public void testFragmentedFrame() {
         EmbeddedChannel encoderChannel = new EmbeddedChannel(
                 ZlibCodecFactory.newZlibEncoder(ZlibWrapper.NONE, 9, 15, 8));
         EmbeddedChannel decoderChannel = new EmbeddedChannel(new PerMessageDeflateDecoder(false));
@@ -106,7 +110,7 @@ public class PerMessageDeflateDecoderTest {
         byte[] payload = new byte[300];
         random.nextBytes(payload);
 
-        encoderChannel.writeOutbound(Unpooled.wrappedBuffer(payload));
+        assertTrue(encoderChannel.writeOutbound(Unpooled.wrappedBuffer(payload)));
         ByteBuf compressedPayload = encoderChannel.readOutbound();
         compressedPayload = compressedPayload.slice(0, compressedPayload.readableBytes() - 4);
 
@@ -121,9 +125,9 @@ public class PerMessageDeflateDecoderTest {
                         compressedPayload.readableBytes() - oneThird * 2));
 
         // execute
-        decoderChannel.writeInbound(compressedFrame1.retain());
+        assertTrue(decoderChannel.writeInbound(compressedFrame1.retain()));
-        decoderChannel.writeInbound(compressedFrame2.retain());
+        assertTrue(decoderChannel.writeInbound(compressedFrame2.retain()));
-        decoderChannel.writeInbound(compressedFrame3);
+        assertTrue(decoderChannel.writeInbound(compressedFrame3));
         BinaryWebSocketFrame uncompressedFrame1 = decoderChannel.readInbound();
         ContinuationWebSocketFrame uncompressedFrame2 = decoderChannel.readInbound();
         ContinuationWebSocketFrame uncompressedFrame3 = decoderChannel.readInbound();
@@ -142,7 +146,7 @@ public class PerMessageDeflateDecoderTest {
 
         byte[] finalPayload = new byte[300];
         finalPayloadWrapped.readBytes(finalPayload);
-        assertTrue(Arrays.equals(finalPayload, payload));
+        assertArrayEquals(finalPayload, payload);
         finalPayloadWrapped.release();
     }
 
@@ -158,9 +162,9 @@ public class PerMessageDeflateDecoderTest {
         byte[] payload2 = new byte[100];
         random.nextBytes(payload2);
 
-        encoderChannel.writeOutbound(Unpooled.wrappedBuffer(payload1));
+        assertTrue(encoderChannel.writeOutbound(Unpooled.wrappedBuffer(payload1)));
         ByteBuf compressedPayload1 = encoderChannel.readOutbound();
-        encoderChannel.writeOutbound(Unpooled.wrappedBuffer(payload2));
+        assertTrue(encoderChannel.writeOutbound(Unpooled.wrappedBuffer(payload2)));
         ByteBuf compressedPayload2 = encoderChannel.readOutbound();
 
         BinaryWebSocketFrame compressedFrame = new BinaryWebSocketFrame(true,
@@ -170,23 +174,140 @@ public class PerMessageDeflateDecoderTest {
                         compressedPayload2.slice(0, compressedPayload2.readableBytes() - 4)));
 
         // execute
-        decoderChannel.writeInbound(compressedFrame);
+        assertTrue(decoderChannel.writeInbound(compressedFrame));
         BinaryWebSocketFrame uncompressedFrame = decoderChannel.readInbound();
 
         // test
         assertNotNull(uncompressedFrame);
         assertNotNull(uncompressedFrame.content());
-        assertTrue(uncompressedFrame instanceof BinaryWebSocketFrame);
         assertEquals(WebSocketExtension.RSV3, uncompressedFrame.rsv());
         assertEquals(200, uncompressedFrame.content().readableBytes());
 
         byte[] finalPayload1 = new byte[100];
         uncompressedFrame.content().readBytes(finalPayload1);
-        assertTrue(Arrays.equals(finalPayload1, payload1));
+        assertArrayEquals(finalPayload1, payload1);
         byte[] finalPayload2 = new byte[100];
         uncompressedFrame.content().readBytes(finalPayload2);
-        assertTrue(Arrays.equals(finalPayload2, payload2));
+        assertArrayEquals(finalPayload2, payload2);
         uncompressedFrame.release();
     }
 
+    @Test
+    public void testDecompressionSkipForBinaryFrame() {
+        EmbeddedChannel encoderChannel = new EmbeddedChannel(
+                ZlibCodecFactory.newZlibEncoder(ZlibWrapper.NONE, 9, 15, 8));
+        EmbeddedChannel decoderChannel = new EmbeddedChannel(new PerMessageDeflateDecoder(false, ALWAYS_SKIP));
+
+        byte[] payload = new byte[300];
+        random.nextBytes(payload);
+
+        assertTrue(encoderChannel.writeOutbound(Unpooled.wrappedBuffer(payload)));
+        ByteBuf compressedPayload = encoderChannel.readOutbound();
+
+        BinaryWebSocketFrame compressedBinaryFrame = new BinaryWebSocketFrame(true, WebSocketExtension.RSV1,
+                                                                              compressedPayload);
+        assertTrue(decoderChannel.writeInbound(compressedBinaryFrame));
+
+        WebSocketFrame inboundFrame = decoderChannel.readInbound();
+
+        assertEquals(WebSocketExtension.RSV1, inboundFrame.rsv());
+        assertEquals(compressedPayload, inboundFrame.content());
+        assertTrue(inboundFrame.release());
+
+        assertTrue(encoderChannel.finishAndReleaseAll());
+        assertFalse(decoderChannel.finish());
+    }
+
+    @Test
+    public void testSelectivityDecompressionSkip() {
+        WebSocketExtensionFilter selectivityDecompressionFilter = new WebSocketExtensionFilter() {
+            @Override
+            public boolean mustSkip(WebSocketFrame frame) {
+                return frame instanceof TextWebSocketFrame && frame.content().readableBytes() < 100;
+            }
+        };
+        EmbeddedChannel encoderChannel = new EmbeddedChannel(
+                ZlibCodecFactory.newZlibEncoder(ZlibWrapper.NONE, 9, 15, 8));
+        EmbeddedChannel decoderChannel = new EmbeddedChannel(
+                new PerMessageDeflateDecoder(false, selectivityDecompressionFilter));
+
+        String textPayload = "compressed payload";
+        byte[] binaryPayload = new byte[300];
+        random.nextBytes(binaryPayload);
+
+        assertTrue(encoderChannel.writeOutbound(Unpooled.wrappedBuffer(textPayload.getBytes(UTF_8))));
+        assertTrue(encoderChannel.writeOutbound(Unpooled.wrappedBuffer(binaryPayload)));
+        ByteBuf compressedTextPayload = encoderChannel.readOutbound();
+        ByteBuf compressedBinaryPayload = encoderChannel.readOutbound();
+
+        TextWebSocketFrame compressedTextFrame = new TextWebSocketFrame(true, WebSocketExtension.RSV1,
+                                                                        compressedTextPayload);
+        BinaryWebSocketFrame compressedBinaryFrame = new BinaryWebSocketFrame(true, WebSocketExtension.RSV1,
+                                                                              compressedBinaryPayload);
+
+        assertTrue(decoderChannel.writeInbound(compressedTextFrame));
+        assertTrue(decoderChannel.writeInbound(compressedBinaryFrame));
+
+        TextWebSocketFrame inboundTextFrame = decoderChannel.readInbound();
+        BinaryWebSocketFrame inboundBinaryFrame = decoderChannel.readInbound();
+
+        assertEquals(WebSocketExtension.RSV1, inboundTextFrame.rsv());
+        assertEquals(compressedTextPayload, inboundTextFrame.content());
+        assertTrue(inboundTextFrame.release());
+
+        assertEquals(0, inboundBinaryFrame.rsv());
+        assertArrayEquals(binaryPayload, ByteBufUtil.getBytes(inboundBinaryFrame.content()));
+        assertTrue(inboundBinaryFrame.release());
+
+        assertTrue(encoderChannel.finishAndReleaseAll());
+        assertFalse(decoderChannel.finish());
+    }
+
+    @Test(expected = DecoderException.class)
+    public void testIllegalStateWhenDecompressionInProgress() {
+        WebSocketExtensionFilter selectivityDecompressionFilter = new WebSocketExtensionFilter() {
+            @Override
+            public boolean mustSkip(WebSocketFrame frame) {
+                return frame.content().readableBytes() < 100;
+            }
+        };
+
+        EmbeddedChannel encoderChannel = new EmbeddedChannel(
+                ZlibCodecFactory.newZlibEncoder(ZlibWrapper.NONE, 9, 15, 8));
+        EmbeddedChannel decoderChannel = new EmbeddedChannel(
+                new PerMessageDeflateDecoder(false, selectivityDecompressionFilter));
+
+        byte[] firstPayload = new byte[200];
+        random.nextBytes(firstPayload);
+
+        byte[] finalPayload = new byte[50];
+        random.nextBytes(finalPayload);
+
+        assertTrue(encoderChannel.writeOutbound(Unpooled.wrappedBuffer(firstPayload)));
+        assertTrue(encoderChannel.writeOutbound(Unpooled.wrappedBuffer(finalPayload)));
+        ByteBuf compressedFirstPayload = encoderChannel.readOutbound();
+        ByteBuf compressedFinalPayload = encoderChannel.readOutbound();
+        assertTrue(encoderChannel.finishAndReleaseAll());
+
+        BinaryWebSocketFrame firstPart = new BinaryWebSocketFrame(false, WebSocketExtension.RSV1,
+                                                                  compressedFirstPayload);
+        ContinuationWebSocketFrame finalPart = new ContinuationWebSocketFrame(true, WebSocketExtension.RSV1,
+                                                                              compressedFinalPayload);
+        assertTrue(decoderChannel.writeInbound(firstPart));
+
+        BinaryWebSocketFrame outboundFirstPart = decoderChannel.readInbound();
+        //first part is decompressed
+        assertEquals(0, outboundFirstPart.rsv());
+        assertArrayEquals(firstPayload, ByteBufUtil.getBytes(outboundFirstPart.content()));
+        assertTrue(outboundFirstPart.release());
+
+        //final part throwing exception
+        try {
+            decoderChannel.writeInbound(finalPart);
+        } finally {
+            assertTrue(finalPart.release());
+            assertFalse(encoderChannel.finishAndReleaseAll());
+        }
+    }
+
 }

codec-http/src/test/java/io/netty/handler/codec/http/websocketx/extensions/compression/PerMessageDeflateEncoderTest.java

@@ -15,20 +15,27 @@
  */
 package io.netty.handler.codec.http.websocketx.extensions.compression;
 
-import static org.junit.Assert.*;
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.EncoderException;
 import io.netty.handler.codec.compression.ZlibCodecFactory;
 import io.netty.handler.codec.compression.ZlibWrapper;
 import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.WebSocketFrame;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtension;
+import io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionFilter;
+import org.junit.Test;
 
 import java.util.Arrays;
 import java.util.Random;
 
-import org.junit.Test;
+import static io.netty.handler.codec.http.websocketx.extensions.WebSocketExtensionFilter.*;
+import static io.netty.util.CharsetUtil.*;
+import static org.junit.Assert.*;
 
 public class PerMessageDeflateEncoderTest {
 
@@ -44,26 +51,25 @@ public class PerMessageDeflateEncoderTest {
         byte[] payload = new byte[300];
         random.nextBytes(payload);
         BinaryWebSocketFrame frame = new BinaryWebSocketFrame(true,
-                WebSocketExtension.RSV3, Unpooled.wrappedBuffer(payload));
+                                                              WebSocketExtension.RSV3, Unpooled.wrappedBuffer(payload));
 
         // execute
-        encoderChannel.writeOutbound(frame);
+        assertTrue(encoderChannel.writeOutbound(frame));
         BinaryWebSocketFrame compressedFrame = encoderChannel.readOutbound();
 
         // test
         assertNotNull(compressedFrame);
         assertNotNull(compressedFrame.content());
-        assertTrue(compressedFrame instanceof BinaryWebSocketFrame);
         assertEquals(WebSocketExtension.RSV1 | WebSocketExtension.RSV3, compressedFrame.rsv());
 
-        decoderChannel.writeInbound(compressedFrame.content());
+        assertTrue(decoderChannel.writeInbound(compressedFrame.content()));
-        decoderChannel.writeInbound(DeflateDecoder.FRAME_TAIL);
+        assertTrue(decoderChannel.writeInbound(Unpooled.wrappedBuffer(DeflateDecoder.FRAME_TAIL)));
         ByteBuf uncompressedPayload = decoderChannel.readInbound();
         assertEquals(300, uncompressedPayload.readableBytes());
 
         byte[] finalPayload = new byte[300];
         uncompressedPayload.readBytes(finalPayload);
-        assertTrue(Arrays.equals(finalPayload, payload));
+        assertArrayEquals(finalPayload, payload);
         uncompressedPayload.release();
     }
 
@@ -76,28 +82,29 @@ public class PerMessageDeflateEncoderTest {
         random.nextBytes(payload);
 
         BinaryWebSocketFrame frame = new BinaryWebSocketFrame(true,
-                WebSocketExtension.RSV3 | WebSocketExtension.RSV1, Unpooled.wrappedBuffer(payload));
+                                                              WebSocketExtension.RSV3 | WebSocketExtension.RSV1,
+                                                              Unpooled.wrappedBuffer(payload));
 
         // execute
-        encoderChannel.writeOutbound(frame);
+        assertTrue(encoderChannel.writeOutbound(frame));
         BinaryWebSocketFrame newFrame = encoderChannel.readOutbound();
 
         // test
         assertNotNull(newFrame);
         assertNotNull(newFrame.content());
-        assertTrue(newFrame instanceof BinaryWebSocketFrame);
         assertEquals(WebSocketExtension.RSV3 | WebSocketExtension.RSV1, newFrame.rsv());
         assertEquals(300, newFrame.content().readableBytes());
 
         byte[] finalPayload = new byte[300];
         newFrame.content().readBytes(finalPayload);
-        assertTrue(Arrays.equals(finalPayload, payload));
+        assertArrayEquals(finalPayload, payload);
         newFrame.release();
     }
 
     @Test
     public void testFramementedFrame() {
-        EmbeddedChannel encoderChannel = new EmbeddedChannel(new PerMessageDeflateEncoder(9, 15, false));
+        EmbeddedChannel encoderChannel = new EmbeddedChannel(new PerMessageDeflateEncoder(9, 15, false,
+                                                                                          NEVER_SKIP));
         EmbeddedChannel decoderChannel = new EmbeddedChannel(
                 ZlibCodecFactory.newZlibDecoder(ZlibWrapper.NONE));
 
@@ -110,16 +117,19 @@ public class PerMessageDeflateEncoderTest {
         random.nextBytes(payload3);
 
         BinaryWebSocketFrame frame1 = new BinaryWebSocketFrame(false,
-                WebSocketExtension.RSV3, Unpooled.wrappedBuffer(payload1));
+                                                               WebSocketExtension.RSV3,
+                                                               Unpooled.wrappedBuffer(payload1));
         ContinuationWebSocketFrame frame2 = new ContinuationWebSocketFrame(false,
-                WebSocketExtension.RSV3, Unpooled.wrappedBuffer(payload2));
+                                                                           WebSocketExtension.RSV3,
+                                                                           Unpooled.wrappedBuffer(payload2));
         ContinuationWebSocketFrame frame3 = new ContinuationWebSocketFrame(true,
-                WebSocketExtension.RSV3, Unpooled.wrappedBuffer(payload3));
+                                                                           WebSocketExtension.RSV3,
+                                                                           Unpooled.wrappedBuffer(payload3));
 
         // execute
-        encoderChannel.writeOutbound(frame1);
+        assertTrue(encoderChannel.writeOutbound(frame1));
-        encoderChannel.writeOutbound(frame2);
+        assertTrue(encoderChannel.writeOutbound(frame2));
-        encoderChannel.writeOutbound(frame3);
+        assertTrue(encoderChannel.writeOutbound(frame3));
         BinaryWebSocketFrame compressedFrame1 = encoderChannel.readOutbound();
         ContinuationWebSocketFrame compressedFrame2 = encoderChannel.readOutbound();
         ContinuationWebSocketFrame compressedFrame3 = encoderChannel.readOutbound();
@@ -135,26 +145,131 @@ public class PerMessageDeflateEncoderTest {
         assertFalse(compressedFrame2.isFinalFragment());
         assertTrue(compressedFrame3.isFinalFragment());
 
-        decoderChannel.writeInbound(compressedFrame1.content());
+        assertTrue(decoderChannel.writeInbound(compressedFrame1.content()));
         ByteBuf uncompressedPayload1 = decoderChannel.readInbound();
         byte[] finalPayload1 = new byte[100];
         uncompressedPayload1.readBytes(finalPayload1);
-        assertTrue(Arrays.equals(finalPayload1, payload1));
+        assertArrayEquals(finalPayload1, payload1);
         uncompressedPayload1.release();
 
-        decoderChannel.writeInbound(compressedFrame2.content());
+        assertTrue(decoderChannel.writeInbound(compressedFrame2.content()));
         ByteBuf uncompressedPayload2 = decoderChannel.readInbound();
         byte[] finalPayload2 = new byte[100];
         uncompressedPayload2.readBytes(finalPayload2);
-        assertTrue(Arrays.equals(finalPayload2, payload2));
+        assertArrayEquals(finalPayload2, payload2);
         uncompressedPayload2.release();
 
-        decoderChannel.writeInbound(compressedFrame3.content());
+        assertTrue(decoderChannel.writeInbound(compressedFrame3.content()));
-        decoderChannel.writeInbound(DeflateDecoder.FRAME_TAIL);
+        assertTrue(decoderChannel.writeInbound(Unpooled.wrappedBuffer(DeflateDecoder.FRAME_TAIL)));
         ByteBuf uncompressedPayload3 = decoderChannel.readInbound();
         byte[] finalPayload3 = new byte[100];
         uncompressedPayload3.readBytes(finalPayload3);
-        assertTrue(Arrays.equals(finalPayload3, payload3));
+        assertArrayEquals(finalPayload3, payload3);
         uncompressedPayload3.release();
     }
+
+    @Test
+    public void testCompressionSkipForBinaryFrame() {
+        EmbeddedChannel encoderChannel = new EmbeddedChannel(new PerMessageDeflateEncoder(9, 15, false,
+                                                                                          ALWAYS_SKIP));
+        byte[] payload = new byte[300];
+        random.nextBytes(payload);
+
+        WebSocketFrame binaryFrame = new BinaryWebSocketFrame(Unpooled.wrappedBuffer(payload));
+
+        assertTrue(encoderChannel.writeOutbound(binaryFrame.copy()));
+        WebSocketFrame outboundFrame = encoderChannel.readOutbound();
+
+        assertEquals(0, outboundFrame.rsv());
+        assertArrayEquals(payload, ByteBufUtil.getBytes(outboundFrame.content()));
+        assertTrue(outboundFrame.release());
+
+        assertFalse(encoderChannel.finish());
+    }
+
+    @Test
+    public void testSelectivityCompressionSkip() {
+        WebSocketExtensionFilter selectivityCompressionFilter = new WebSocketExtensionFilter() {
+            @Override
+            public boolean mustSkip(WebSocketFrame frame) {
+                return  (frame instanceof TextWebSocketFrame || frame instanceof BinaryWebSocketFrame)
+                    && frame.content().readableBytes() < 100;
+            }
+        };
+        EmbeddedChannel encoderChannel = new EmbeddedChannel(
+                new PerMessageDeflateEncoder(9, 15, false, selectivityCompressionFilter));
+        EmbeddedChannel decoderChannel = new EmbeddedChannel(
+                ZlibCodecFactory.newZlibDecoder(ZlibWrapper.NONE));
+
+        String textPayload = "not compressed payload";
+        byte[] binaryPayload = new byte[101];
+        random.nextBytes(binaryPayload);
+
+        WebSocketFrame textFrame = new TextWebSocketFrame(textPayload);
+        BinaryWebSocketFrame binaryFrame = new BinaryWebSocketFrame(Unpooled.wrappedBuffer(binaryPayload));
+
+        assertTrue(encoderChannel.writeOutbound(textFrame));
+        assertTrue(encoderChannel.writeOutbound(binaryFrame));
+
+        WebSocketFrame outboundTextFrame = encoderChannel.readOutbound();
+
+        //compression skipped for textFrame
+        assertEquals(0, outboundTextFrame.rsv());
+        assertEquals(textPayload, outboundTextFrame.content().toString(UTF_8));
+        assertTrue(outboundTextFrame.release());
+
+        WebSocketFrame outboundBinaryFrame = encoderChannel.readOutbound();
+
+        //compression not skipped for binaryFrame
+        assertEquals(WebSocketExtension.RSV1, outboundBinaryFrame.rsv());
+
+        assertTrue(decoderChannel.writeInbound(outboundBinaryFrame.content().retain()));
+        ByteBuf uncompressedBinaryPayload = decoderChannel.readInbound();
+
+        assertArrayEquals(binaryPayload, ByteBufUtil.getBytes(uncompressedBinaryPayload));
+
+        assertTrue(outboundBinaryFrame.release());
+        assertTrue(uncompressedBinaryPayload.release());
+
+        assertFalse(encoderChannel.finish());
+        assertFalse(decoderChannel.finish());
+    }
+
+    @Test(expected = EncoderException.class)
+    public void testIllegalStateWhenCompressionInProgress() {
+        WebSocketExtensionFilter selectivityCompressionFilter = new WebSocketExtensionFilter() {
+            @Override
+            public boolean mustSkip(WebSocketFrame frame) {
+                return frame.content().readableBytes() < 100;
+            }
+        };
+        EmbeddedChannel encoderChannel = new EmbeddedChannel(
+                new PerMessageDeflateEncoder(9, 15, false, selectivityCompressionFilter));
+
+        byte[] firstPayload = new byte[200];
+        random.nextBytes(firstPayload);
+
+        byte[] finalPayload = new byte[90];
+        random.nextBytes(finalPayload);
+
+        BinaryWebSocketFrame firstPart = new BinaryWebSocketFrame(false, 0, Unpooled.wrappedBuffer(firstPayload));
+        ContinuationWebSocketFrame finalPart = new ContinuationWebSocketFrame(true, 0,
+                                                                              Unpooled.wrappedBuffer(finalPayload));
+        assertTrue(encoderChannel.writeOutbound(firstPart));
+
+        BinaryWebSocketFrame outboundFirstPart = encoderChannel.readOutbound();
+        //first part is compressed
+        assertEquals(WebSocketExtension.RSV1, outboundFirstPart.rsv());
+        assertFalse(Arrays.equals(firstPayload, ByteBufUtil.getBytes(outboundFirstPart.content())));
+        assertTrue(outboundFirstPart.release());
+
+        //final part throwing exception
+        try {
+            encoderChannel.writeOutbound(finalPart);
+        } finally {
+            assertTrue(finalPart.release());
+            assertFalse(encoderChannel.finishAndReleaseAll());
+        }
+    }
+
 }