DirectByteBufInStringOutPayload extends AbstractInOutPayload { @Override public String toString() { return result; }  @Override String toString(); static final Cumulator COMPOSITE_CUMULATOR; static final Cumulator MERGE_CUMULATOR; }
@Test public void testScaleOut() throws Exception { DirectByteBufInStringOutPayload payload = new DirectByteBufInStringOutPayload(); String randomStr = randomString(); ByteBuf byteBuf = directByteBuf(randomStr.length()); randomStr += randomString(2<<11); byteBuf.writeBytes(randomStr.getBytes()); payload.startInput(); payload.in(byteBuf); payload.endInput(); Assert.assertEquals(randomStr, payload.toString()); }
@Test public void testContinuouslyInput() throws Exception { DirectByteBufInStringOutPayload payload = new DirectByteBufInStringOutPayload(); StringBuilder randomStr = new StringBuilder(); payload.startInput(); ByteBufAllocator allocator = new PooledByteBufAllocator(); for(int i = 0; i < 100; i ++) { String delta = randomString(100); ByteBuf byteBuf = allocator.directBuffer(delta.length()); byteBuf.writeBytes(delta.getBytes()); randomStr.append(delta); payload.in(byteBuf); } payload.endInput(); Assert.assertEquals(randomStr.toString(), payload.toString()); sleep(5000); }
@Test public void testContinuouslyWithLargeInput() throws Exception { DirectByteBufInStringOutPayload payload = new DirectByteBufInStringOutPayload(); StringBuilder randomStr = new StringBuilder(); payload.startInput(); for(int i = 0; i < 10; i ++) { String delta = randomString(1024); ByteBuf byteBuf = directByteBuf(delta.length()); byteBuf.writeBytes(delta.getBytes()); randomStr.append(delta); payload.in(byteBuf); byteBuf.release(); } payload.endInput(); Assert.assertEquals(randomStr.toString(), payload.toString()); sleep(5000); }