fix byte buffer leak

Author: brainlag

src/main/java/com/github/brainlag/nsq/Connection.java

@@ -40,7 +40,7 @@ public class Connection {
     private final EventLoopGroup eventLoopGroup;
     private final NSQConfig config;
 
-    public static final long HEARTBEAT_MAX_INTERVAL = 1L*60L*1000L;//default one minute
+    public static final long HEARTBEAT_MAX_INTERVAL = 1L * 60L * 1000L;//default one minute
     private volatile AtomicReference<Long> lastHeartbeatSuccess = new AtomicReference<Long>(System.currentTimeMillis());
 
 
@@ -97,8 +97,8 @@ public boolean isRequestInProgress() {
         return requests.size() > 0;
     }
 
-    public boolean isHeartbeatStatusOK(){
-        if(System.currentTimeMillis() - lastHeartbeatSuccess.get() > HEARTBEAT_MAX_INTERVAL){
+    public boolean isHeartbeatStatusOK() {
+        if (System.currentTimeMillis() - lastHeartbeatSuccess.get() > HEARTBEAT_MAX_INTERVAL) {
             return false;
         }
         return true;

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

@@ -4,44 +4,46 @@
 import io.netty.buffer.Unpooled;
 
 public class MessageFrame extends NSQFrame {
-	private long timestamp;
-	private int attempts;
-	private byte[] messageId = new byte[16];
-	private byte[] messageBody;
-	
-	@Override
-	public void setData(byte[] bytes) {
-		//parse the bytes
-		super.setData(bytes);
+    private long timestamp;
+    private int attempts;
+    private byte[] messageId = new byte[16];
+    private byte[] messageBody;
+
+    @Override
+    public void setData(byte[] bytes) {
+        //parse the bytes
+        super.setData(bytes);
 
         ByteBuf buf = Unpooled.wrappedBuffer(bytes);
-		timestamp = buf.readLong();
-		attempts = buf.readShort();
-		buf.readBytes(messageId);
-		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() {
-		return timestamp;
-	}
-
-	public int getAttempts() {
-		return attempts;
-	}
-
-	public byte[] getMessageId() {
-		return messageId;
-	}
-
-	public byte[] getMessageBody() {
-		return messageBody;
-	}
+        timestamp = buf.readLong();
+        attempts = buf.readShort();
+        buf.readBytes(messageId);
+        ByteBuf messageBodyBuf = buf.readBytes(buf.readableBytes());
+        if (messageBodyBuf.hasArray()) {
+            messageBody = messageBodyBuf.array();
+        } else {
+            byte[] array = new byte[messageBodyBuf.readableBytes()];
+            messageBodyBuf.readBytes(array);
+            messageBody = array;
+        }
+        buf.release();
+        messageBodyBuf.release();
+    }
+
+    public long getTimestamp() {
+        return timestamp;
+    }
+
+    public int getAttempts() {
+        return attempts;
+    }
+
+    public byte[] getMessageId() {
+        return messageId;
+    }
+
+    public byte[] getMessageBody() {
+        return messageBody;
+    }
 
 }

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

@@ -9,31 +9,32 @@
 
 public class NSQDecoder extends MessageToMessageDecoder<ByteBuf> {
 
-	private int size;
-	private NSQFrame frame;
+    private int size;
+    private NSQFrame frame;
 
-	 public NSQDecoder() {
-	 }
+    public NSQDecoder() {
+    }
 
     @Override
     protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
-		size = in.readInt();
-		int id = in.readInt();
-		frame = NSQFrame.instance(id);
-		if (frame == null) {
-			//uhh, bad response from server..  what should we do?
-			throw new Exception("Bad frame id from server (" + id + ").  disconnect!");
-		}
-		frame.setSize(size);
-		ByteBuf bytes = in.readBytes(frame.getSize() - 4); //subtract 4 because the frame id is included
-		if (bytes.hasArray()) {
-			frame.setData(bytes.array());
-		} else {
-			byte[] array = new byte[bytes.readableBytes()];
-			bytes.readBytes(array);
-			frame.setData(array);
-		}
-		out.add(frame);
-	}
+        size = in.readInt();
+        int id = in.readInt();
+        frame = NSQFrame.instance(id);
+        if (frame == null) {
+            //uhh, bad response from server..  what should we do?
+            throw new Exception("Bad frame id from server (" + id + ").  disconnect!");
+        }
+        frame.setSize(size);
+        ByteBuf bytes = in.readBytes(frame.getSize() - 4); //subtract 4 because the frame id is included
+        if (bytes.hasArray()) {
+            frame.setData(bytes.array());
+        } else {
+            byte[] array = new byte[bytes.readableBytes()];
+            bytes.readBytes(array);
+            frame.setData(array);
+        }
+        out.add(frame);
+        bytes.release();
+    }
 
 }