@Test
public void resetDelegatesToUnderlyingItem() {

    // given
    SimpleInputBufferPooledObjectOps ops = new SimpleInputBufferPooledObjectOps(
            HeapByteBufferAllocator.INSTANCE,
            TEST_BUFFER_SIZE
    );

    ItemSource<SimpleInputBuffer> result = spy(ops.createItemSource(source -> {}));
    SimpleInputBuffer simpleInputBuffer = mock(SimpleInputBuffer.class);
    when(result.getSource()).thenReturn(simpleInputBuffer);

    // when
    ops.reset(result);

    // then
    verify(simpleInputBuffer).reset();

}
 

@Test
public void closeReleasesBufferOnlyOnce() throws IOException {

    // given
    ItemSource itemSource = mock(ItemSource.class);
    when(itemSource.getSource()).thenReturn(mock(SimpleInputBuffer.class));

    InputStream inputStream = new ItemSourceContentInputStream(itemSource);

    // when
    inputStream.close();
    inputStream.close();

    // then
    verify(itemSource, times(1)).release();

}
 

@Test
public void resetDelegatesToSimpleInputBuffer() {

    // given
    SimpleInputBufferPooledObjectOps pooledObjectOps = spy(createDefaultTestObject());
    SimpleInputBuffer inputBuffer = mock(SimpleInputBuffer.class);

    ItemSource<SimpleInputBuffer> itemSourceMock = mock(ItemSource.class);
    when(pooledObjectOps.createItemSource(any())).thenReturn(itemSourceMock);

    ItemSource<SimpleInputBuffer> itemSource = pooledObjectOps.createItemSource(null);
    when(itemSourceMock.getSource()).thenReturn(inputBuffer);

    // when
    pooledObjectOps.reset(itemSource);

    // then
    Mockito.verify(inputBuffer, times(1)).reset();

}
 

@Test
public void purgeHasNoSideEffects() {

    // given
    SimpleInputBufferPooledObjectOps pooledObjectOps = spy(createDefaultTestObject());
    SimpleInputBuffer inputBuffer = mock(SimpleInputBuffer.class);

    ItemSource<SimpleInputBuffer> itemSourceMock = mock(ItemSource.class);
    when(pooledObjectOps.createItemSource(any())).thenReturn(itemSourceMock);

    ItemSource<SimpleInputBuffer> itemSource = pooledObjectOps.createItemSource(null);
    when(itemSourceMock.getSource()).thenReturn(inputBuffer);

    // when
    pooledObjectOps.purge(itemSource);

    // then
    verifyZeroInteractions(itemSourceMock);
    verifyZeroInteractions(inputBuffer);

}
 

public static GenericItemSourcePool<SimpleInputBuffer> createDefaultTestGenericItemSourcePool(
        int initialSize,
        boolean monitored,
        ResizePolicy resizePolicy
) {

    SimpleInputBufferPooledObjectOps pooledObjectOps = new SimpleInputBufferPooledObjectOps(
            HeapByteBufferAllocator.INSTANCE,
            DEFAULT_TEST_ITEM_SIZE_IN_BYTES);

    return new GenericItemSourcePool<>(
            DEFAULT_TEST_ITEM_POOL_NAME,
            pooledObjectOps,
            resizePolicy,
            DEFAULT_TEST_RESIZE_TIMEOUT,
            monitored,
            DEFAULT_TEST_MONITOR_TASK_INTERVAL,
            initialSize
    );
}
 

@Test
public void onEntityEnclosedSetsResponseInputStream() throws IOException {

    // given
    GenericItemSourcePool<SimpleInputBuffer> itemSourcePool = createDefaultTestGenericItemSourcePool(
                    GenericItemSourcePoolTest.DEFAULT_TEST_INITIAL_POOL_SIZE,
                    false
            );

    PoolingAsyncResponseConsumer consumer = createDefaultTestObject(itemSourcePool);
    consumer.onResponseReceived(mock(HttpResponse.class));

    HttpEntity httpEntity = mock(HttpEntity.class);

    // when
    consumer.onEntityEnclosed(httpEntity, ContentType.create("application/json"));

    // then
    HttpResponse response = consumer.buildResult(null);
    verify(response, times(1)).setEntity(any());

}
 

@Test
public void onEntityEnclosedPoolsTheBufferOnce() throws IOException, PoolResourceException {

    // given
    GenericItemSourcePool<SimpleInputBuffer> itemSourcePool = spy(createDefaultTestGenericItemSourcePool(
            GenericItemSourcePoolTest.DEFAULT_TEST_INITIAL_POOL_SIZE,
            false
    ));

    PoolingAsyncResponseConsumer consumer = spy(createDefaultTestObject(itemSourcePool));
    consumer.onResponseReceived(mock(HttpResponse.class));

    HttpEntity httpEntity = mock(HttpEntity.class);

    // when
    consumer.onEntityEnclosed(httpEntity, ContentType.create("application/json"));
    consumer.onEntityEnclosed(httpEntity, ContentType.create("application/json"));

    // then
    verify(consumer, times(1)).getPooled();

}
 

@Test
public void releaseResourcesNullifiesTheResponse() throws IOException {

    // given
    GenericItemSourcePool<SimpleInputBuffer> itemSourcePool = createDefaultTestGenericItemSourcePool(
            GenericItemSourcePoolTest.DEFAULT_TEST_INITIAL_POOL_SIZE,
            false
    );

    PoolingAsyncResponseConsumer consumer = createDefaultTestObject(itemSourcePool);
    consumer.onResponseReceived(mock(HttpResponse.class));

    HttpEntity httpEntity = mock(HttpEntity.class);
    consumer.onEntityEnclosed(httpEntity, ContentType.create("application/json"));

    HttpResponse before = consumer.buildResult(null);
    assertNotNull(before);

    // when
    consumer.releaseResources();

    // then
    HttpResponse response = consumer.buildResult(null);
    assertNull(response);

}
 

protected void onEntityEnclosed(HttpEntity entity, ContentType contentType) throws IOException {
    long len = entity.getContentLength();
    if (len > 2147483647L) {
        throw new ContentTooLongException("Entity content is too long: " + len);
    } else {
        if (len < 0L) {
            len = 4096L;
        }

        this.buf = new SimpleInputBuffer((int)len, new HeapByteBufferAllocator());
        this.httpResponse.setEntity(new ContentBufferEntity(entity, this.buf));
    }
}
 

private GenericItemSourcePool<SimpleInputBuffer> createPool() {
    GenericItemSourcePool<SimpleInputBuffer> bufferPool = new GenericItemSourcePool<>(
            "hc-responseBufferPool",
            new SimpleInputBufferPooledObjectOps(
                    HeapByteBufferAllocator.INSTANCE,
                    pooledResponseBuffersSizeInBytes
            ),
            new UnlimitedResizePolicy.Builder().withResizeFactor(0.5).build(),
            1000L,
            false,
            30000,
            maxTotalConnections
    );
    return bufferPool;
}
 

ItemSource<SimpleInputBuffer> getPooled() throws IOException {
    try {
        return itemSourcePool.getPooled();
    } catch (PoolResourceException e) {
        throw new IOException("Unable get pooled response buffer: " + e.getMessage());
    }
}
 

@Test
public void releaseDelegatesToReleaseCallback() {

    // given
    SimpleInputBuffer buffer = mock(SimpleInputBuffer.class);
    ReleaseCallback releaseCallback = mock(ReleaseCallback.class);
    ItemSource<SimpleInputBuffer> itemSource = new InputBufferItemSource(buffer, releaseCallback);

    // when
    itemSource.release();

    // then
    verify(releaseCallback, times(1)).completed(eq(itemSource));

}
 

@Override
protected void onEntityEnclosed(final HttpEntity entity,
        final ContentType contentType) throws IOException {
    long len = entity.getContentLength();
    if (len > Integer.MAX_VALUE) {
        throw new ContentTooLongException("Entity content is too long: "
                + len);
    }
    if (len < 0) {
        len = BUFFER_SIZE;
    }
    this.buf = new SimpleInputBuffer((int) len,
            new HeapByteBufferAllocator());
    this.httpResponse.setEntity(new ContentBufferEntity(entity, this.buf));
}
 

public ItemSourceContentInputStream(ItemSource<SimpleInputBuffer> buffer) {
    super(buffer.getSource());
    this.buffer = buffer;
}
 

public PoolingAsyncResponseConsumer(ItemSourcePool<SimpleInputBuffer> bufferPool) {
    this.itemSourcePool = bufferPool;
}
 

public PoolingAsyncResponseConsumerFactory(GenericItemSourcePool<SimpleInputBuffer> pool) {
    this.pool = pool;
}
 

@Override
public ItemSource<SimpleInputBuffer> createItemSource(ReleaseCallback<SimpleInputBuffer> releaseCallback) {
    SimpleInputBuffer buffer = new SimpleInputBuffer(bufferSizeInBytes, byteBufAllocator);
    return new InputBufferItemSource(buffer, releaseCallback);
}
 

@Override
public void reset(ItemSource<SimpleInputBuffer> pooled) {
    pooled.getSource().reset();
}
 

@Override
public boolean purge(ItemSource<SimpleInputBuffer> pooled) {
    return true;
}
 

public InputBufferItemSource(SimpleInputBuffer source, ReleaseCallback<SimpleInputBuffer> releaseCallback) {
    this.source = source;
    this.releaseCallback = releaseCallback;
}
 

@Override
public SimpleInputBuffer getSource() {
    return source;
}
 

public static GenericItemSourcePool<SimpleInputBuffer> createDefaultTestGenericItemSourcePool(int initialSize, boolean monitored) {
    ResizePolicy resizePolicy = UnlimitedResizePolicy.newBuilder().build();
    return createDefaultTestGenericItemSourcePool(initialSize, monitored, resizePolicy);
}
 

private PoolingAsyncResponseConsumer createDefaultTestObject(ItemSourcePool<SimpleInputBuffer> itemSourcePool) {
    return new PoolingAsyncResponseConsumer(itemSourcePool);
}
 

private PoolingAsyncResponseConsumer createDefaultTestObject() {
    ItemSourcePool<SimpleInputBuffer> itemSourcePool = mock(ItemSourcePool.class);
    return createDefaultTestObject(itemSourcePool);
}
 

@Test
public void getSourceReturnsSameItemSourceInstance() {

    // given
    SimpleInputBuffer expected = mock(SimpleInputBuffer.class);

    // when
    ItemSource<SimpleInputBuffer> itemSource = new InputBufferItemSource(expected, null);

    // then
    Assert.assertTrue(expected == itemSource.getSource());

}
 

@Test
public void onContentReceivedPassedDecoderToBuffer() throws IOException, PoolResourceException {

    // given
    ItemSourcePool<SimpleInputBuffer> itemSourcePool = mock(ItemSourcePool.class);

    ItemSource<SimpleInputBuffer> itemSource = mock(ItemSource.class);
    when(itemSourcePool.getPooled()).thenReturn(itemSource);

    SimpleInputBuffer buffer = mock(SimpleInputBuffer.class);
    when(itemSource.getSource()).thenReturn(buffer);

    PoolingAsyncResponseConsumer consumer = createDefaultTestObject(itemSourcePool);
    consumer.onResponseReceived(mock(HttpResponse.class));

    HttpEntity httpEntity = mock(HttpEntity.class);
    consumer.onEntityEnclosed(httpEntity, ContentType.create("application/json"));

    ContentDecoder contentDecoder = mock(ContentDecoder.class);

    // when
    consumer.onContentReceived(contentDecoder, mock(IOControl.class));

    // then
    verify(buffer, times(1)).consumeContent(eq(contentDecoder));

}