@Test
public void testBufferStaging() 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);
	// --------------------------------------------------------------------
	Envelope[] envelopes = nextEnvelopes(3, true);
	ByteBuf buf = encode(ch, envelopes);
	when(this.bufferProvider.registerBufferAvailabilityListener(Matchers.<BufferAvailabilityListener>anyObject()))
			.thenReturn(BufferAvailabilityRegistration.SUCCEEDED_REGISTERED);
	Buffer buffer = allocBuffer(envelopes[2].getBuffer().size());
	when(this.bufferProvider.requestBuffer(anyInt()))
			.thenReturn(null, null, buffer, null);
	// --------------------------------------------------------------------
	// slices: [0] => full envelope, [1] => half envelope, [2] => remaining half + full envelope
	ByteBuf[] slices = slice(buf,
			OutboundEnvelopeEncoder.HEADER_SIZE + envelopes[0].getBuffer().size(),
			OutboundEnvelopeEncoder.HEADER_SIZE + envelopes[1].getBuffer().size() / 2);
	// 1. no buffer available, incoming slice contains all data
	int refCount = slices[0].refCnt();
	decodeAndVerify(ch, slices[0]);
	Assert.assertEquals(refCount + 1, slices[0].refCnt());
	Assert.assertFalse(ch.config().isAutoRead());
	// notify of available buffer (=> bufferAvailable() callback does return a buffer
	// of the current network buffer size; the decoder needs to adjust its size to the
	// requested size
	decoder.bufferAvailable(allocBuffer(envelopes[0].getBuffer().size() * 2));
	ch.runPendingTasks();
	Assert.assertEquals(refCount - 1, slices[0].refCnt());
	Assert.assertTrue(ch.config().isAutoRead());
	decodeAndVerify(ch, envelopes[0]);
	// 2. no buffer available, incoming slice does NOT contain all data
	refCount = slices[1].refCnt();
	decodeAndVerify(ch, slices[1]);
	Assert.assertEquals(refCount + 1, slices[1].refCnt());
	Assert.assertFalse(ch.config().isAutoRead());
	decoder.bufferAvailable(allocBuffer());
	ch.runPendingTasks();
	Assert.assertEquals(refCount - 1, slices[1].refCnt());
	Assert.assertTrue(ch.config().isAutoRead());
	decodeAndVerify(ch);
	// 3. buffer available
	refCount = slices[2].refCnt();
	decodeAndVerify(ch, slices[2], envelopes[1], envelopes[2]);
	Assert.assertEquals(refCount - 1, slices[2].refCnt());
	Assert.assertTrue(ch.config().isAutoRead());
	Assert.assertEquals(1, buf.refCnt());
	buf.release();
}