Merge pull request #31 from dpapworth/fix_netty

Fix ByteBuf handling for newer versions of Netty

Author: brainlag

pom.xml

@@ -142,7 +142,7 @@
     <dependency>
       <groupId>io.netty</groupId>
       <artifactId>netty-all</artifactId>
-      <version>4.0.34.Final</version>
+      <version>4.0.39.Final</version>
     </dependency>
 
     <dependency>
@@ -173,7 +173,7 @@
     <dependency>
       <groupId>io.netty</groupId>
       <artifactId>netty-tcnative</artifactId>
-      <version>1.1.33.Fork11</version>
+      <version>1.1.33.Fork18</version>
       <classifier>${os.detected.classifier}</classifier>
       <scope>test</scope>
     </dependency>

src/main/java/com/github/brainlag/nsq/frames/MessageFrame.java

@@ -18,7 +18,14 @@ public void setData(byte[] bytes) {
 		timestamp = buf.readLong();
 		attempts = buf.readShort();
 		buf.readBytes(messageId);
-		messageBody = buf.readBytes(buf.readableBytes()).array();
+		ByteBuf messageBodyBuf = buf.readBytes(buf.readableBytes());
+		if (messageBodyBuf.hasArray()) {
+			messageBody = messageBodyBuf.array();
+		} else {
+			byte[] array = new byte[messageBodyBuf.readableBytes()];
+			messageBodyBuf.readBytes(array);
+			messageBody = array;
+		}
 	}
 
 	public long getTimestamp() {

src/main/java/com/github/brainlag/nsq/netty/NSQDecoder.java

@@ -26,7 +26,13 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
 		}
 		frame.setSize(size);
 		ByteBuf bytes = in.readBytes(frame.getSize() - 4); //subtract 4 because the frame id is included
-		frame.setData(bytes.array());
+		if (bytes.hasArray()) {
+			frame.setData(bytes.array());
+		} else {
+			byte[] array = new byte[bytes.readableBytes()];
+			bytes.readBytes(array);
+			frame.setData(array);
+		}
 		out.add(frame);
 	}