NioPlainEngine implements NioFilter { @Override public ByteBuffer readAtLeast(SocketChannel channel, int bytes, ByteBuffer wrappedBuffer) throws IOException { ByteBuffer buffer = wrappedBuffer; Assert.assertTrue(buffer.capacity() - lastProcessedPosition >= bytes); buffer.limit(buffer.capacity()); buffer.position(lastReadPosition); while (buffer.position() < (lastProcessedPosition + bytes)) { int amountRead = channel.read(buffer); if (amountRead < 0) { throw new EOFException(); } } lastReadPosition = buffer.position(); buffer.limit(lastProcessedPosition + bytes); buffer.position(lastProcessedPosition); lastProcessedPosition += bytes; return buffer; } NioPlainEngine(BufferPool bufferPool); @Override ByteBuffer wrap(ByteBuffer buffer); @Override ByteBuffer unwrap(ByteBuffer wrappedBuffer); @Override ByteBuffer ensureWrappedCapacity(int amount, ByteBuffer wrappedBuffer,
      BufferPool.BufferType bufferType); @Override ByteBuffer readAtLeast(SocketChannel channel, int bytes, ByteBuffer wrappedBuffer); void doneReading(ByteBuffer unwrappedBuffer); @Override ByteBuffer getUnwrappedBuffer(ByteBuffer wrappedBuffer);  }
@Test public void readAtLeast() throws Exception { final int amountToRead = 150; final int individualRead = 60; final int preexistingBytes = 10; ByteBuffer wrappedBuffer = ByteBuffer.allocate(1000); SocketChannel mockChannel = mock(SocketChannel.class); when(mockChannel.read(any(ByteBuffer.class))).thenAnswer(new Answer<Integer>() { @Override public Integer answer(InvocationOnMock invocation) throws Throwable { ByteBuffer buffer = invocation.getArgument(0); buffer.position(buffer.position() + individualRead); return individualRead; } }); nioEngine.lastReadPosition = 10; ByteBuffer data = nioEngine.readAtLeast(mockChannel, amountToRead, wrappedBuffer); verify(mockChannel, times(3)).read(isA(ByteBuffer.class)); assertThat(data.position()).isEqualTo(0); assertThat(data.limit()).isEqualTo(amountToRead); assertThat(nioEngine.lastReadPosition).isEqualTo(individualRead * 3 + preexistingBytes); assertThat(nioEngine.lastProcessedPosition).isEqualTo(amountToRead); data = nioEngine.readAtLeast(mockChannel, amountToRead, wrappedBuffer); verify(mockChannel, times(5)).read(any(ByteBuffer.class)); assertThat(data.position()).isEqualTo(amountToRead); assertThat(data.limit()).isEqualTo(amountToRead * 2); assertThat(nioEngine.lastReadPosition).isEqualTo(individualRead * 5 + preexistingBytes); assertThat(nioEngine.lastProcessedPosition).isEqualTo(amountToRead * 2); }
@Test(expected = EOFException.class) public void readAtLeastThrowsEOFException() throws Exception { final int amountToRead = 150; ByteBuffer wrappedBuffer = ByteBuffer.allocate(1000); SocketChannel mockChannel = mock(SocketChannel.class); when(mockChannel.read(any(ByteBuffer.class))).thenReturn(-1); nioEngine.lastReadPosition = 10; nioEngine.readAtLeast(mockChannel, amountToRead, wrappedBuffer); }