@Test public void testProduceDataWithBuffering() throws Exception { final AsyncEntityProducer producer = new ChunkCharAsyncEntityProducer( 256, 5, ContentType.TEXT_PLAIN, "this", " and that", "all", " sorts of stuff"); final WritableByteChannelMock byteChannel = new WritableByteChannelMock(1024); final DataStreamChannel streamChannel = new BasicDataStreamChannel(byteChannel); producer.produce(streamChannel); Assert.assertTrue(byteChannel.isOpen()); Assert.assertEquals("", byteChannel.dump(StandardCharsets.US_ASCII)); producer.produce(streamChannel); Assert.assertTrue(byteChannel.isOpen()); Assert.assertEquals("this and that", byteChannel.dump(StandardCharsets.US_ASCII)); producer.produce(streamChannel); Assert.assertTrue(byteChannel.isOpen()); Assert.assertEquals("", byteChannel.dump(StandardCharsets.US_ASCII)); producer.produce(streamChannel); Assert.assertFalse(byteChannel.isOpen()); Assert.assertEquals("all sorts of stuff", byteChannel.dump(StandardCharsets.US_ASCII)); }
@Override public final void produce(final DataStreamChannel channel) throws IOException { synchronized (bytebuf) { if (state == State.ACTIVE) { produceData(new StreamChannel<CharBuffer>() { @Override public int write(final CharBuffer src) throws IOException { Args.notNull(src, "Buffer"); synchronized (bytebuf) { return writeData(channel, src); } } @Override public void endStream() throws IOException { synchronized (bytebuf) { streamEnd(channel); } } }); } if (state == State.FLUSHING) { final CoderResult result = charsetEncoder.flush(bytebuf); if (result.isError()) { result.throwException(); } else if (result.isOverflow()) { flush(channel); } else if (result.isUnderflow()) { flush(channel); if (bytebuf.position() == 0) { state = State.END_STREAM; channel.endStream(); } } } } }
AbstractCharAsyncEntityProducer implements AsyncEntityProducer { @Override public final void produce(final DataStreamChannel channel) throws IOException { synchronized (bytebuf) { if (state == State.ACTIVE) { produceData(new StreamChannel<CharBuffer>() { @Override public int write(final CharBuffer src) throws IOException { Args.notNull(src, "Buffer"); synchronized (bytebuf) { return writeData(channel, src); } } @Override public void endStream() throws IOException { synchronized (bytebuf) { streamEnd(channel); } } }); } if (state == State.FLUSHING) { final CoderResult result = charsetEncoder.flush(bytebuf); if (result.isError()) { result.throwException(); } else if (result.isOverflow()) { flush(channel); } else if (result.isUnderflow()) { flush(channel); if (bytebuf.position() == 0) { state = State.END_STREAM; channel.endStream(); } } } } } }
AbstractCharAsyncEntityProducer implements AsyncEntityProducer { @Override public final void produce(final DataStreamChannel channel) throws IOException { synchronized (bytebuf) { if (state == State.ACTIVE) { produceData(new StreamChannel<CharBuffer>() { @Override public int write(final CharBuffer src) throws IOException { Args.notNull(src, "Buffer"); synchronized (bytebuf) { return writeData(channel, src); } } @Override public void endStream() throws IOException { synchronized (bytebuf) { streamEnd(channel); } } }); } if (state == State.FLUSHING) { final CoderResult result = charsetEncoder.flush(bytebuf); if (result.isError()) { result.throwException(); } else if (result.isOverflow()) { flush(channel); } else if (result.isUnderflow()) { flush(channel); if (bytebuf.position() == 0) { state = State.END_STREAM; channel.endStream(); } } } } } AbstractCharAsyncEntityProducer(
            final int bufferSize,
            final int fragmentSizeHint,
            final ContentType contentType); }
AbstractCharAsyncEntityProducer implements AsyncEntityProducer { @Override public final void produce(final DataStreamChannel channel) throws IOException { synchronized (bytebuf) { if (state == State.ACTIVE) { produceData(new StreamChannel<CharBuffer>() { @Override public int write(final CharBuffer src) throws IOException { Args.notNull(src, "Buffer"); synchronized (bytebuf) { return writeData(channel, src); } } @Override public void endStream() throws IOException { synchronized (bytebuf) { streamEnd(channel); } } }); } if (state == State.FLUSHING) { final CoderResult result = charsetEncoder.flush(bytebuf); if (result.isError()) { result.throwException(); } else if (result.isOverflow()) { flush(channel); } else if (result.isUnderflow()) { flush(channel); if (bytebuf.position() == 0) { state = State.END_STREAM; channel.endStream(); } } } } } AbstractCharAsyncEntityProducer(
            final int bufferSize,
            final int fragmentSizeHint,
            final ContentType contentType); @Override final String getContentType(); @Override String getContentEncoding(); @Override long getContentLength(); @Override boolean isChunked(); @Override Set<String> getTrailerNames(); @Override final int available(); @Override final void produce(final DataStreamChannel channel); @Override void releaseResources(); }
AbstractCharAsyncEntityProducer implements AsyncEntityProducer { @Override public final void produce(final DataStreamChannel channel) throws IOException { synchronized (bytebuf) { if (state == State.ACTIVE) { produceData(new StreamChannel<CharBuffer>() { @Override public int write(final CharBuffer src) throws IOException { Args.notNull(src, "Buffer"); synchronized (bytebuf) { return writeData(channel, src); } } @Override public void endStream() throws IOException { synchronized (bytebuf) { streamEnd(channel); } } }); } if (state == State.FLUSHING) { final CoderResult result = charsetEncoder.flush(bytebuf); if (result.isError()) { result.throwException(); } else if (result.isOverflow()) { flush(channel); } else if (result.isUnderflow()) { flush(channel); if (bytebuf.position() == 0) { state = State.END_STREAM; channel.endStream(); } } } } } AbstractCharAsyncEntityProducer(
            final int bufferSize,
            final int fragmentSizeHint,
            final ContentType contentType); @Override final String getContentType(); @Override String getContentEncoding(); @Override long getContentLength(); @Override boolean isChunked(); @Override Set<String> getTrailerNames(); @Override final int available(); @Override final void produce(final DataStreamChannel channel); @Override void releaseResources();  }