package com.github.dockerjava.netty.handler;

import java.io.OutputStream;

import com.github.dockerjava.api.model.Frame;
import com.github.dockerjava.api.model.StreamType;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class DockerRawStreamHandler extends SimpleChannelInboundHandler<ByteBuf> {

	private static final int HEADER_SIZE = 8;

	private final ByteBuf rawBuffer = Unpooled.buffer(1000);

	private OutputStream stdout, stderr;

	private byte[] header = new byte[HEADER_SIZE];

	private int headerCnt = 0;

	private int payloadCnt = 0;

	public DockerRawStreamHandler(OutputStream stdout, OutputStream stderr) {
		this.stdout = stdout;
		this.stderr = stderr;
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {

		// System.out.println("msg: " + msg.readableBytes());
		rawBuffer.writeBytes(msg.copy());
		// System.out.println("readableBytes: " + rawBuffer.readableBytes());

		Frame frame = null;

		do {
			if (frame != null) {
				switch (frame.getStreamType()) {
				case STDOUT:
					if (stdout != null) {
						stdout.write(frame.getPayload());
						stdout.flush();
					}
					break;
				case STDERR:
					if (stderr != null) {
						stderr.write(frame.getPayload());
						stderr.flush();
					}
					break;
				}
				headerCnt = 0;
				payloadCnt = 0;
			}
		} while ((frame = decode()) != null);

	}

	private int read(byte[] buf, int offset, int length) {
		length = Math.min(rawBuffer.readableBytes(), length);
		rawBuffer.readBytes(buf, offset, length);
		return length;
	}

	private Frame decode() {
		if (headerCnt < HEADER_SIZE) {

			int headerCount = read(header, headerCnt, HEADER_SIZE - headerCnt);

			if (headerCount == 0) {
				return null;
			}
			headerCnt += headerCount;

			if (headerCnt < HEADER_SIZE)
				return null;
		}

		StreamType streamType = streamType(header[0]);

		int payloadSize = ((header[4] & 0xff) << 24) + ((header[5] & 0xff) << 16) + ((header[6] & 0xff) << 8)
				+ (header[7] & 0xff);

		byte[] payload = new byte[payloadSize];

		int count = read(payload, payloadCnt, payloadSize - payloadCnt);

		if (count == 0)
			return null;

		payloadCnt += count;

		if (payloadCnt < payloadSize)
			return null;

		return new Frame(streamType, payload);

	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		cause.printStackTrace();
		ctx.close();
	}

	private static StreamType streamType(byte streamType) {
		switch (streamType) {
		case 0:
			return StreamType.STDIN;
		case 1:
			return StreamType.STDOUT;
		case 2:
			return StreamType.STDERR;
		default:
			return StreamType.RAW;
		}
	}

}

@@ -631,24 +631,65 @@
 aderCnt = 0;
+%0A%09%0A%09private byte%5B%5D payload = new byte%5B0%5D;
 %0A%0A%09private i
@@ -941,151 +941,47 @@
 %0A%0A%09%09
-// System.out.println(%22msg: %22 + msg.readableBytes());%0A%09%09rawBuffer.writeBytes(msg.copy());%0A%09%09// System.out.println(%22readableBytes: %22 + rawBuffer
+rawBuffer.writeBytes(msg.copy(), 0, msg
 .rea
@@ -1026,16 +1026,41 @@
 %0A%09%09do %7B%0A
+%09%09%09frame = decode();%0A%09%09%09%0A
 %09%09%09if (f
@@ -1362,22 +1362,90 @@
 %09break;%0A
+%09%09%09%09default:%0A%09%09%09%09%09System.err.println(%22unknown stream type%22);%09%09%0A
 %09%09%09%09%7D%0A
+%09%09%09%09%0A
 %09%09%09%09head
@@ -1479,16 +1479,21 @@
  0;%0A%09%09%09%7D
+ %0A%09%09%09
 %0A%09%09%7D whi
@@ -1499,27 +1499,14 @@
 ile 
-(
 (frame
- = decode())
  != 
@@ -2142,23 +2142,39 @@
 ff);%0A%0A%09%09
-byte%5B%5D 
+if(payloadCnt == 0)%0A%09%09%09
 payload 

Fix regression in decode algorithm for docker stream