Merge pull request #451 from NiteshKant/0.4.x

Upgrade netty to 4.1.0.Beta8

Author: NiteshKant

gradle.properties

@@ -13,5 +13,5 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-netty_version=4.1.0.Beta7
+netty_version=4.1.0.Beta8
 slf4j_version=1.7.6

rxnetty/src/main/java/io/reactivex/netty/protocol/http/sse/ServerSentEventDecoder.java

@@ -16,9 +16,9 @@
 package io.reactivex.netty.protocol.http.sse;
 
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufProcessor;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.util.ByteProcessor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -38,36 +38,36 @@ public class ServerSentEventDecoder extends ByteToMessageDecoder {
     private static final char[] DATA_FIELD_NAME = "data".toCharArray();
     private static final char[] ID_FIELD_NAME = "id".toCharArray();
 
-    protected static final ByteBufProcessor SKIP_TILL_LINE_DELIMITER_PROCESSOR = new ByteBufProcessor() {
+    protected static final ByteProcessor SKIP_TILL_LINE_DELIMITER_PROCESSOR = new ByteProcessor() {
         @Override
         public boolean process(byte value) throws Exception {
             return !isLineDelimiter((char) value);
         }
     };
 
-    protected static final ByteBufProcessor SKIP_LINE_DELIMITERS_AND_SPACES_PROCESSOR = new ByteBufProcessor() {
+    protected static final ByteProcessor SKIP_LINE_DELIMITERS_AND_SPACES_PROCESSOR = new ByteProcessor() {
         @Override
         public boolean process(byte value) throws Exception {
             return isLineDelimiter((char) value) || (char) value == ' ';
         }
     };
 
-    protected static final ByteBufProcessor SKIP_COLON_AND_WHITE_SPACE_PROCESSOR = new ByteBufProcessor() {
+    protected static final ByteProcessor SKIP_COLON_AND_WHITE_SPACE_PROCESSOR = new ByteProcessor() {
         @Override
         public boolean process(byte value) throws Exception {
             char valueChar = (char) value;
             return valueChar == ':' || valueChar == ' ';
         }
     };
 
-    protected static final ByteBufProcessor SCAN_COLON_PROCESSOR = new ByteBufProcessor() {
+    protected static final ByteProcessor SCAN_COLON_PROCESSOR = new ByteProcessor() {
         @Override
         public boolean process(byte value) throws Exception {
             return (char) value != ':';
         }
     };
 
-    protected static final ByteBufProcessor SCAN_EOL_PROCESSOR = new ByteBufProcessor() {
+    protected static final ByteProcessor SCAN_EOL_PROCESSOR = new ByteProcessor() {
         @Override
         public boolean process(byte value) throws Exception {
             return !isLineDelimiter((char) value);
@@ -340,7 +340,7 @@ private static boolean skipTillEOL(ByteBuf in) {
         return skipTillMatching(in, SKIP_TILL_LINE_DELIMITER_PROCESSOR);
     }
 
-    protected static boolean skipTillMatching(ByteBuf byteBuf, ByteBufProcessor processor) {
+    protected static boolean skipTillMatching(ByteBuf byteBuf, ByteProcessor processor) {
         final int lastIndexProcessed = byteBuf.forEachByte(processor);
         if (-1 == lastIndexProcessed) {
             byteBuf.readerIndex(byteBuf.readerIndex() + byteBuf.readableBytes()); // If all the remaining bytes are to be ignored, discard the buffer.

rxnetty/src/main/java/io/reactivex/netty/protocol/http/sse/ServerSentEventEncoder.java

@@ -16,10 +16,10 @@
 package io.reactivex.netty.protocol.http.sse;
 
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufProcessor;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.MessageToByteEncoder;
+import io.netty.util.ByteProcessor;
 
 /**
  * An encoder to convert {@link ServerSentEvent} to a {@link io.netty.buffer.ByteBuf}
@@ -67,7 +67,7 @@ protected void encode(ChannelHandlerContext ctx, ServerSentEvent serverSentEvent
         if (splitSseData) {
             while (content.isReadable()) { // Scan the buffer and split on new line into multiple data lines.
                 final int readerIndexAtStart = content.readerIndex();
-                int newLineIndex = content.forEachByte(new ByteBufProcessor() {
+                int newLineIndex = content.forEachByte(new ByteProcessor() {
                     @Override
                     public boolean process(byte value) throws Exception {
                         return (char) value != '\n';