ByteArrayOutputStreamPayload extends AbstractInOutPayload { public byte[] getBytes(){ int currentPos = pos.get(); byte []dst = new byte[currentPos]; System.arraycopy(data, 0, dst, 0, currentPos); return dst; } ByteArrayOutputStreamPayload();  ByteArrayOutputStreamPayload(int INIT_SIZE);  ByteArrayOutputStreamPayload(String message); @Override void doStartInput(); @Override int doIn(ByteBuf byteBuf); @Override long doOut(WritableByteChannel writableByteChannel); byte[] getBytes(); @Override String toString();  }
@Test public void testInout() throws Exception { ByteArrayOutputStreamPayload payload = new ByteArrayOutputStreamPayload(); String randomStr = randomString(); ByteBuf byteBuf = directByteBuf(randomStr.length()); byteBuf.writeBytes(randomStr.getBytes()); payload.startInput(); payload.in(byteBuf); payload.endInput(); final ByteBuf result = directByteBuf(randomStr.length()); payload.startOutput(); long wroteLength = payload.out(new WritableByteChannel() { @Override public boolean isOpen() { return false; } @Override public void close() throws IOException { } @Override public int write(ByteBuffer src) throws IOException { int readable = result.readableBytes(); result.writeBytes(src); return result.readableBytes() - readable; } }); payload.endOutput(); Assert.assertEquals(randomStr.length(), wroteLength); byte []resultArray = new byte[(int) wroteLength]; result.readBytes(resultArray); Assert.assertEquals(randomStr, new String(resultArray)); }
@Test public void testNewHeap() throws Exception { final MemoryPrinter memoryPrinter = new MemoryPrinter(scheduled); memoryPrinter.printMemory(); final int length = 1 << 10; int concurrentCount = 10; final CountDownLatch latch = new CountDownLatch(concurrentCount); final ByteBuf byteBuf = directByteBuf(length); byteBuf.writeBytes(randomString(length).getBytes()); byte []dst = new byte[length]; byteBuf.readBytes(dst); memoryPrinter.printMemory(); for(int i=0;i<concurrentCount;i++){ Thread current = new Thread( new AbstractExceptionLogTask() { @Override protected void doRun() throws Exception { try{ byteBuf.readerIndex(0); ByteArrayOutputStream baous = new ByteArrayOutputStream(); byteBuf.readBytes(baous, length); }finally{ latch.countDown(); } } }); current.start(); memoryPrinter.printMemory(); } latch.await(); }
@Test public void testScaleOut() throws Exception { ByteArrayOutputStreamPayload payload = new ByteArrayOutputStreamPayload(); String randomStr = randomString(); ByteBuf byteBuf = directByteBuf(randomStr.length()); randomStr += randomString(2<<11); byteBuf.writeBytes(randomStr.getBytes()); payload.startInput(); payload.in(byteBuf); payload.endInput(); }