MultiplexedStreamHandler extends AbstractStreamHandler { @Nullable @Override public StdioInputStream getNextStreamFragment() throws IOException { if (previousStreamFragment != null) { previousStreamFragment.exhaustAndClose(); } int n; int headerOffset = 0; while ((n = inputStream.read(headerBuffer, headerOffset, headerBuffer.length - headerOffset)) != -1) { headerOffset += n; assert headerOffset <= headerBuffer.length; if (headerOffset == headerBuffer.length) { ByteBuffer bb = ByteBuffer.wrap(headerBuffer); bb.order(ByteOrder.LITTLE_ENDIAN); StdioType type; try { type = StdioType.fromStreamType(DockerCloudUtils.toUnsignedLong(bb.getInt())); } catch (IllegalArgumentException e) { throw new IOException("Invalid stream content.", e); } bb.order(ByteOrder.BIG_ENDIAN); long fragmentLength = DockerCloudUtils.toUnsignedLong(bb.getInt()); assert !bb.hasRemaining() : "Header not fully parsed."; CappedInputStream streamFragment = new CappedInputStream(inputStream, fragmentLength); previousStreamFragment = streamFragment; return new StdioInputStream(streamFragment, type); } } if (headerOffset != 0) { LOG.warn("Underflow while reading stream header."); } return null; } MultiplexedStreamHandler(Closeable closeHandle, InputStream inputStream, OutputStream outputStream); @Nullable @Override StdioInputStream getNextStreamFragment();  }
@Test public void demultiplexing() throws IOException { inputStream = new TestInputStream(multiplexedStream); StreamHandler handler = createHandler(); StdioInputStream streamFragment = handler.getNextStreamFragment(); assertThat(streamFragment).isNotNull(); assertThat(streamFragment.getType()).isSameAs(StdioType.STDIN); assertThat(DockerCloudUtils.readUTF8String(streamFragment)).isEqualTo(STDIN_MSG); streamFragment = handler.getNextStreamFragment(); assertThat(streamFragment).isNotNull(); assertThat(streamFragment.getType()).isSameAs(StdioType.STDOUT); assertThat(DockerCloudUtils.readUTF8String(streamFragment)).isEqualTo(STDOUT_MSG); streamFragment = handler.getNextStreamFragment(); assertThat(streamFragment).isNotNull(); assertThat(streamFragment.getType()).isSameAs(StdioType.STDERR); assertThat(DockerCloudUtils.readUTF8String(streamFragment)).isEqualTo(STDERR_MSG); }