org.apache.http.nio.ContentDecoder#org.apache.http.nio.IOControl源码实例Demo

public void produceContent(final ContentEncoder enc, final IOControl ioc) throws IOException {
    if (content != null) {
        if (buffer == null) {
            if (content.getTempFile() == null) {
                buffer = ByteBuffer.wrap(content.getBytes());
            } else {
                fis = content.getInputStream();
                chan = (fis instanceof FileInputStream)
                    ? ((FileInputStream)fis).getChannel() : Channels.newChannel(fis);
                buffer = ByteBuffer.allocate(8 * 1024);
            }
        }
        int i = -1;
        ((Buffer)buffer).rewind();
        if (buffer.hasRemaining() && chan != null) {
            i = chan.read(buffer);
            buffer.flip();
        }
        enc.write(buffer);
        if (!buffer.hasRemaining() && i == -1) {
            enc.complete();
        }
    } else {
        buf.produceContent(enc, ioc);
    }
}
 

@Override
protected void onByteReceived(ByteBuffer buf, IOControl ioctrl) throws IOException {
	if (session == null)
		session = task.createSession();

	received += buf.remaining();
	session.receiveData(buf);
	downloadCallback.updateProgress(received, contextLength);
}
 

public int produceContent(final ContentEncoder encoder, final IOControl ioc) throws IOException {
    if (this.shutdown) {
        return -1;
    }
    this.lock.lock();
    try {
        this.ioctrl = ioc;
        setOutputMode();
        int bytesWritten = 0;
        if (largeWrapper != null || super.hasData()) {
            if (!this.buffer.hasRemaining() && largeWrapper != null) {
                bytesWritten = encoder.write(largeWrapper);
            } else {
                bytesWritten = encoder.write(this.buffer);
            }
            if (encoder.isCompleted()) {
                this.endOfStream = true;
            }
        }
        if ((largeWrapper == null || !largeWrapper.hasRemaining()) && !super.hasData()) {
            // No more buffered content
            // If at the end of the stream, terminate
            if (this.endOfStream && !encoder.isCompleted()) {
                encoder.complete();
            }
            if (!this.endOfStream && this.ioctrl != null) {
                // suspend output events
                this.ioctrl.suspendOutput();
            }
        }
        // no need to signal if the large wrapper is present and has data remaining
        if (largeWrapper == null || !largeWrapper.hasRemaining()) {
            this.condition.signalAll();
        }
        return bytesWritten;
    } finally {
        this.lock.unlock();
    }
}
 

@Override
public void consumeContent(final ContentDecoder dec, final IOControl ioc) throws IOException {
    // Only consume content when the work was accepted by the work queue
    if (outstream.retrySetHttpResponse(response)) {
        buf.consumeContent(dec, ioc);
    }
}
 

@Override
public void produceContent(ContentEncoder encoder, IOControl ioctrl) throws IOException {
    delegate.produceContent(encoder, ioctrl);
}
 

protected void onContentReceived(ContentDecoder decoder, IOControl ioControl) throws IOException {
    this.buf.consumeContent(decoder);
}
 

@Override
protected void onContentReceived(
        final ContentDecoder decoder, final IOControl ioctrl) throws IOException {
    this.buffer.getSource().consumeContent(decoder);
}
 

@Override
protected void onCharReceived(final CharBuffer buf, final IOControl ioctrl) throws IOException {
    while (buf.hasRemaining()) {
        System.out.print(buf.get());
    }
}
 

@Override
protected void onCharReceived(final CharBuffer buf, final IOControl ioctrl) throws IOException {
    while (buf.hasRemaining()) {
        System.out.print(buf.get());
    }
}
 

@Override
protected void onCharReceived(final CharBuffer buf, final IOControl ioctrl) throws IOException {
	this.url.result += buf.toString();
}
 

@Override
protected void onContentReceived(final ContentDecoder decoder,
        final IOControl ioctrl) throws IOException {
    Preconditions.checkNotNull(this.buf, "Content buffer should not be null.");
    this.buf.consumeContent(decoder);
}
 

@Override
public void consumeContent(final ContentDecoder decoder, final IOControl control) throws IOException {
    delegate().consumeContent(decoder, control);
}
 

@Override
public void consumeContent(ContentDecoder decoder, IOControl ioctrl) throws IOException {
    consumer.consumeContent(decoder, ioctrl);
}
 

public int consumeContent(final ContentDecoder decoder, final IOControl ioc) throws IOException {
    if (this.shutdown) {
        //something bad happened, we need to shutdown the connection
        //as we're not going to read the data at all and we
        //don't want to keep getting read notices and such
        ioc.shutdown();
        return -1;
    }
    this.lock.lock();
    try {
        this.ioctrl = ioc;
        setInputMode();
        int totalRead = 0;
        int bytesRead;
        if (waitingBuffer != null && this.buffer.position() == 0) {
            while ((bytesRead = decoder.read(this.waitingBuffer)) > 0) {
                totalRead += bytesRead;
            }
        }
        //read more
        while ((bytesRead = decoder.read(this.buffer)) > 0) {
            totalRead += bytesRead;
        }
        if (bytesRead == -1 || decoder.isCompleted()) {
            this.endOfStream = true;
        }
        if (!this.buffer.hasRemaining() && this.ioctrl != null && !this.endOfStream) {
            this.ioctrl.suspendInput();
        }
        this.condition.signalAll();

        if (totalRead > 0) {
            return totalRead;
        }
        if (this.endOfStream) {
            return -1;
        }
        return 0;
    } finally {
        this.lock.unlock();
    }
}
 

@Override public void produceContent(ContentEncoder encoder, IOControl io) throws IOException {
  requestProducer.produceContent(encoder, io);
}
 

@Override public void consumeContent(ContentDecoder decoder, IOControl ioctrl)
  throws IOException {
  responseConsumer.consumeContent(decoder, ioctrl);
}
 

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

}