@Test
public void testEncodeDecodeRandomEnvelopes() throws Exception {

	final InboundEnvelopeDecoder decoder = new InboundEnvelopeDecoder(this.bufferProviderBroker);
	final EmbeddedChannel ch = new EmbeddedChannel(
			new OutboundEnvelopeEncoder(), decoder);
	when(this.bufferProviderBroker.getBufferProvider(anyJobId(), anyChannelId()))
			.thenReturn(this.bufferProvider);
	when(this.bufferProvider.requestBuffer(anyInt())).thenAnswer(new Answer<Object>() {
		@Override
		public Object answer(InvocationOnMock invocation) throws Throwable {
			// fulfill the buffer request with the requested size
			return allocBuffer((Integer) invocation.getArguments()[0]);
		}
	});
	Random randomAnswerSource = new Random(RANDOM_SEED);
	RandomBufferRequestAnswer randomBufferRequestAnswer = new RandomBufferRequestAnswer(randomAnswerSource);
	RandomBufferAvailabilityRegistrationAnswer randomBufferAvailabilityRegistrationAnswer =
			new RandomBufferAvailabilityRegistrationAnswer(randomAnswerSource, randomBufferRequestAnswer);
	when(this.bufferProvider.requestBuffer(anyInt())).thenAnswer(randomBufferRequestAnswer);
	when(this.bufferProvider.registerBufferAvailabilityListener(Matchers.<BufferAvailabilityListener>anyObject()))
			.thenAnswer(randomBufferAvailabilityRegistrationAnswer);
	// --------------------------------------------------------------------
	Envelope[] envelopes = nextRandomEnvelopes(1024);
	ByteBuf buf = encode(ch, envelopes);
	ByteBuf[] slices = randomSlices(buf);
	for (ByteBuf slice : slices) {
		int refCount = slice.refCnt();
		ch.writeInbound(slice);
		// registered BufferAvailabilityListener => call bufferAvailable(buffer)
		while (randomBufferAvailabilityRegistrationAnswer.isRegistered()) {
			randomBufferAvailabilityRegistrationAnswer.unregister();
			Assert.assertFalse(ch.config().isAutoRead());
			Assert.assertEquals(refCount + 1, slice.refCnt());
			// return a buffer of max size => decoder needs to limit buffer size
			decoder.bufferAvailable(allocBuffer(MAX_BUFFER_SIZE));
			ch.runPendingTasks();
		}
		Assert.assertEquals(refCount - 1, slice.refCnt());
		Assert.assertTrue(ch.config().isAutoRead());
	}
	Envelope[] expected = randomBufferAvailabilityRegistrationAnswer.removeSkippedEnvelopes(envelopes);
	decodeAndVerify(ch, expected);
	Assert.assertEquals(1, buf.refCnt());
	buf.release();
}