BufferPool { ByteBuffer expandReadBufferIfNeeded(BufferType type, ByteBuffer existing, int desiredCapacity) { if (existing.capacity() >= desiredCapacity) { if (existing.position() > 0) { existing.compact(); existing.flip(); } return existing; } ByteBuffer newBuffer; if (existing.isDirect()) { newBuffer = acquireDirectBuffer(type, desiredCapacity); } else { newBuffer = acquireNonDirectBuffer(type, desiredCapacity); } newBuffer.clear(); newBuffer.put(existing); newBuffer.flip(); releaseBuffer(type, existing); return newBuffer; } BufferPool(DMStats stats); ByteBuffer acquireDirectSenderBuffer(int size); ByteBuffer acquireDirectReceiveBuffer(int size); ByteBuffer acquireNonDirectSenderBuffer(int size); ByteBuffer acquireNonDirectReceiveBuffer(int size); void releaseSenderBuffer(ByteBuffer bb); void releaseReceiveBuffer(ByteBuffer bb); @VisibleForTesting ByteBuffer getPoolableBuffer(ByteBuffer buffer); static final boolean useDirectBuffers; }
@Test public void bufferPositionAndLimitForReadAreCorrectAfterExpansion() throws Exception { ByteBuffer buffer = ByteBuffer.allocate(33842); buffer.position(7); buffer.limit(16384); ByteBuffer newBuffer = bufferPool.expandReadBufferIfNeeded(BufferPool.BufferType.UNTRACKED, buffer, 40899); assertThat(newBuffer.capacity()).isGreaterThanOrEqualTo(40899); assertThat(newBuffer.position()).isEqualTo(0); assertThat(newBuffer.limit()).isEqualTo(16384 - 7); }