@Test
public void createsItemSource() {

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

    // when
    ItemSource result = ops.createItemSource(mock(ReleaseCallback.class));

    // then
    assertNotNull(result);

}
 

@Test
public void purgeDoesNotAffectItemSource() {

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

    ItemSource result = spy(ops.createItemSource(mock(ReleaseCallback.class)));

    // when
    ops.purge(result);

    // then
    verifyNoMoreInteractions(result);

}
 

@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 metricsSupplierReturnsNull() {

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

    Supplier<String> metricsSupplier = ops.createMetricsSupplier();

    // when
    String result = metricsSupplier.get();

    // then
    assertNull(result);
}
 

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
    );
}
 

public AsyncWrappedOutputStream(Message message,
                                boolean needToCacheRequest,
                                boolean isChunking,
                                int chunkThreshold,
                                String conduitName,
                                URI uri) {
    super(message,
          needToCacheRequest,
          isChunking,
          chunkThreshold,
          conduitName,
          uri);
    csPolicy = getClient(message);
    entity = message.get(CXFHttpRequest.class);
    basicEntity = (BasicHttpEntity)entity.getEntity();
    basicEntity.setChunked(isChunking);
    HeapByteBufferAllocator allocator = new HeapByteBufferAllocator();
    int bufSize = csPolicy.getChunkLength() > 0 ? csPolicy.getChunkLength() : 16320;
    inbuf = new SharedInputBuffer(bufSize, allocator);
    outbuf = new SharedOutputBuffer(bufSize, allocator);
    isAsync = outMessage != null && outMessage.getExchange() != null
        && !outMessage.getExchange().isSynchronous();
}
 

@Override
public boolean startListener(int port, String name, InboundProcessorParams params) {
    log.info("Starting HL7 Inbound Endpoint on port " + port);

    Map<String, Object> parameters = new HashMap<String, Object>();
    parameters.put(MLLPConstants.INBOUND_PARAMS, params);
    parameters.put(MLLPConstants.INBOUND_HL7_BUFFER_FACTORY,
                   new BufferFactory(8 * 1024, new HeapByteBufferAllocator(), 1024));
    validateParameters(params, parameters);

    HL7Processor hl7Processor = new HL7Processor(parameters);
    parameters.put(MLLPConstants.HL7_REQ_PROC, hl7Processor);

    return InboundHL7IOReactor.bind(port, hl7Processor);
}
 

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;
}
 

@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));
}
 

protected void setupNewConnection(String newURL) throws IOException {
    httpResponse = null;
    isAsync = outMessage != null && outMessage.getExchange() != null
        && !outMessage.getExchange().isSynchronous();
    exception = null;
    connectionFuture = null;
    session = null;
    sslState = null;
    sslURL = null;

    //reset the buffers
    HeapByteBufferAllocator allocator = new HeapByteBufferAllocator();
    int bufSize = csPolicy.getChunkLength() > 0 ? csPolicy.getChunkLength() : 16320;
    inbuf = new SharedInputBuffer(bufSize, allocator);
    outbuf = new SharedOutputBuffer(bufSize, allocator);
    try {
        if (defaultAddress.getString().equals(newURL)) {
            setupConnection(outMessage, defaultAddress, csPolicy);
        } else {
            Address address = new Address(newURL);
            this.url = address.getURI();
            setupConnection(outMessage, address, csPolicy);
        }
        entity = outMessage.get(CXFHttpRequest.class);
        basicEntity = (BasicHttpEntity)entity.getEntity();
        entity.setOutputStream(this);
    } catch (URISyntaxException e) {
        throw new IOException(e);
    }
}
 

public static SimpleInputBufferPooledObjectOps createDefaultTestObject() {
    return new SimpleInputBufferPooledObjectOps(HeapByteBufferAllocator.INSTANCE, DEFAULT_TEST_ITEM_SIZE_IN_BYTES);
}