@Test public void testProduceDataWithBuffering2() throws Exception { final AsyncEntityProducer producer = new ChunkByteAsyncEntityProducer( 5, ContentType.TEXT_PLAIN, new byte[] { '1' }, new byte[] { '2' }, new byte[] { '3' }, new byte[] { '4', '5' }, new byte[] { '6' }, new byte[] { '7', '8' }, new byte[] { '9', '0' }); 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("", byteChannel.dump(StandardCharsets.US_ASCII)); producer.produce(streamChannel); Assert.assertTrue(byteChannel.isOpen()); Assert.assertEquals("", byteChannel.dump(StandardCharsets.US_ASCII)); producer.produce(streamChannel); Assert.assertTrue(byteChannel.isOpen()); Assert.assertEquals("12345", byteChannel.dump(StandardCharsets.US_ASCII)); producer.produce(streamChannel); Assert.assertTrue(byteChannel.isOpen()); Assert.assertEquals("", 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("67890", byteChannel.dump(StandardCharsets.US_ASCII)); }
@Override public final void produce(final DataStreamChannel channel) throws IOException { synchronized (bytebuf) { if (state == State.ACTIVE) { produceData(new StreamChannel<ByteBuffer>() { @Override public int write(final ByteBuffer 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) { flush(channel); if (bytebuf.position() == 0) { state = State.END_STREAM; channel.endStream(); } } } }
AbstractBinAsyncEntityProducer implements AsyncEntityProducer { @Override public final void produce(final DataStreamChannel channel) throws IOException { synchronized (bytebuf) { if (state == State.ACTIVE) { produceData(new StreamChannel<ByteBuffer>() { @Override public int write(final ByteBuffer 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) { flush(channel); if (bytebuf.position() == 0) { state = State.END_STREAM; channel.endStream(); } } } } }
AbstractBinAsyncEntityProducer implements AsyncEntityProducer { @Override public final void produce(final DataStreamChannel channel) throws IOException { synchronized (bytebuf) { if (state == State.ACTIVE) { produceData(new StreamChannel<ByteBuffer>() { @Override public int write(final ByteBuffer 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) { flush(channel); if (bytebuf.position() == 0) { state = State.END_STREAM; channel.endStream(); } } } } AbstractBinAsyncEntityProducer(final int fragmentSizeHint, final ContentType contentType); }
AbstractBinAsyncEntityProducer implements AsyncEntityProducer { @Override public final void produce(final DataStreamChannel channel) throws IOException { synchronized (bytebuf) { if (state == State.ACTIVE) { produceData(new StreamChannel<ByteBuffer>() { @Override public int write(final ByteBuffer 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) { flush(channel); if (bytebuf.position() == 0) { state = State.END_STREAM; channel.endStream(); } } } } AbstractBinAsyncEntityProducer(final int fragmentSizeHint, final ContentType contentType); @Override final String getContentType(); @Override String getContentEncoding(); @Override boolean isChunked(); @Override Set<String> getTrailerNames(); @Override long getContentLength(); @Override final int available(); @Override final void produce(final DataStreamChannel channel); @Override void releaseResources(); }
AbstractBinAsyncEntityProducer implements AsyncEntityProducer { @Override public final void produce(final DataStreamChannel channel) throws IOException { synchronized (bytebuf) { if (state == State.ACTIVE) { produceData(new StreamChannel<ByteBuffer>() { @Override public int write(final ByteBuffer 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) { flush(channel); if (bytebuf.position() == 0) { state = State.END_STREAM; channel.endStream(); } } } } AbstractBinAsyncEntityProducer(final int fragmentSizeHint, final ContentType contentType); @Override final String getContentType(); @Override String getContentEncoding(); @Override boolean isChunked(); @Override Set<String> getTrailerNames(); @Override long getContentLength(); @Override final int available(); @Override final void produce(final DataStreamChannel channel); @Override void releaseResources();  }