LengthDelimitedDecoder extends AbstractContentDecoder implements FileContentDecoder { @Override public long transfer( final FileChannel dst, final long position, final long count) throws IOException { if (dst == null) { return 0; } if (isCompleted()) { return -1; } final int chunk = (int) Math.min((this.contentLength - this.len), Integer.MAX_VALUE); final long bytesRead; if (this.buffer.hasData()) { final int maxLen = Math.min(chunk, this.buffer.length()); dst.position(position); bytesRead = this.buffer.read(dst, count < maxLen ? (int)count : maxLen); } else { if (this.channel.isOpen()) { if (position > dst.size()) { throw new IOException(String.format("Position past end of file [%d > %d]", position, dst.size())); } bytesRead = dst.transferFrom(this.channel, position, count < chunk ? count : chunk); } else { bytesRead = -1; } if (bytesRead > 0) { this.metrics.incrementBytesTransferred(bytesRead); } } if (bytesRead == -1) { setCompleted(); if (this.len < this.contentLength) { throw new ConnectionClosedException( "Premature end of Content-Length delimited message body (expected: %d; received: %d)", this.contentLength, this.len); } } this.len += bytesRead; if (this.len >= this.contentLength) { this.completed = true; } return bytesRead; } LengthDelimitedDecoder(
            final ReadableByteChannel channel,
            final SessionInputBuffer buffer,
            final BasicHttpTransportMetrics metrics,
            final long contentLength); @Override int read(final ByteBuffer dst); @Override long transfer(
            final FileChannel dst,
            final long position,
            final long count); @Override String toString();  }
@Test public void testBasicDecodingFile() throws Exception { final ReadableByteChannel channel = new ReadableByteChannelMock( new String[] {"stuff; ", "more stuff; ", "a lot more stuff!!!"}, StandardCharsets.US_ASCII); final SessionInputBuffer inbuf = new SessionInputBufferImpl(1024, 256, 0, StandardCharsets.US_ASCII); final BasicHttpTransportMetrics metrics = new BasicHttpTransportMetrics(); final LengthDelimitedDecoder decoder = new LengthDelimitedDecoder( channel, inbuf, metrics, 36); createTempFile(); final RandomAccessFile testfile = new RandomAccessFile(this.tmpfile, "rw"); try { final FileChannel fchannel = testfile.getChannel(); long pos = 0; while (!decoder.isCompleted()) { final long bytesRead = decoder.transfer(fchannel, pos, 10); if (bytesRead > 0) { pos += bytesRead; } } } finally { testfile.close(); } Assert.assertEquals(this.tmpfile.length(), metrics.getBytesTransferred()); Assert.assertEquals("stuff; more stuff; a lot more stuff!", CodecTestUtils.readFromFile(this.tmpfile)); }
@Test public void testDecodingFileWithBufferedSessionData() throws Exception { final ReadableByteChannel channel = new ReadableByteChannelMock( new String[] {"stuff; ", "more stuff; ", "a lot more stuff!!!"}, StandardCharsets.US_ASCII); final SessionInputBuffer inbuf = new SessionInputBufferImpl(1024, 256, 0, StandardCharsets.US_ASCII); final BasicHttpTransportMetrics metrics = new BasicHttpTransportMetrics(); final LengthDelimitedDecoder decoder = new LengthDelimitedDecoder( channel, inbuf, metrics, 36); final int i = inbuf.fill(channel); Assert.assertEquals(7, i); createTempFile(); final RandomAccessFile testfile = new RandomAccessFile(this.tmpfile, "rw"); try { final FileChannel fchannel = testfile.getChannel(); long pos = 0; while (!decoder.isCompleted()) { final long bytesRead = decoder.transfer(fchannel, pos, 10); if (bytesRead > 0) { pos += bytesRead; } } } finally { testfile.close(); } Assert.assertEquals(this.tmpfile.length() - 7, metrics.getBytesTransferred()); Assert.assertEquals("stuff; more stuff; a lot more stuff!", CodecTestUtils.readFromFile(this.tmpfile)); }
@Test public void testDecodingFileWithOffsetAndBufferedSessionData() throws Exception { final ReadableByteChannel channel = new ReadableByteChannelMock( new String[] {"stuff; ", "more stuff; ", "a lot more stuff!"}, StandardCharsets.US_ASCII); final SessionInputBuffer inbuf = new SessionInputBufferImpl(1024, 256, 0, StandardCharsets.US_ASCII); final BasicHttpTransportMetrics metrics = new BasicHttpTransportMetrics(); final LengthDelimitedDecoder decoder = new LengthDelimitedDecoder( channel, inbuf, metrics, 36); final int i = inbuf.fill(channel); Assert.assertEquals(7, i); final byte[] beginning = "beginning; ".getBytes(StandardCharsets.US_ASCII); createTempFile(); RandomAccessFile testfile = new RandomAccessFile(this.tmpfile, "rw"); try { testfile.write(beginning); } finally { testfile.close(); } testfile = new RandomAccessFile(this.tmpfile, "rw"); try { final FileChannel fchannel = testfile.getChannel(); long pos = beginning.length; while (!decoder.isCompleted()) { if(testfile.length() < pos) { testfile.setLength(pos); } final long bytesRead = decoder.transfer(fchannel, pos, 10); if (bytesRead > 0) { pos += bytesRead; } } } finally { testfile.close(); } Assert.assertEquals(this.tmpfile.length() - 7 - beginning.length, metrics.getBytesTransferred()); Assert.assertEquals("beginning; stuff; more stuff; a lot more stuff!", CodecTestUtils.readFromFile(this.tmpfile)); }
@Test public void testDecodingFileWithLimit() throws Exception { final ReadableByteChannel channel = new ReadableByteChannelMock( new String[] {"stuff; more stuff; ", "a lot more stuff!!!"}, StandardCharsets.US_ASCII); final SessionInputBuffer inbuf = new SessionInputBufferImpl(1024, 256, 0, StandardCharsets.US_ASCII); final BasicHttpTransportMetrics metrics = new BasicHttpTransportMetrics(); final LengthDelimitedDecoder decoder = new LengthDelimitedDecoder(channel, inbuf, metrics, 36); final int i = inbuf.fill(channel); Assert.assertEquals(19, i); createTempFile(); final RandomAccessFile testfile = new RandomAccessFile(this.tmpfile, "rw"); try { final FileChannel fchannel = testfile.getChannel(); long pos = 0; long bytesRead = decoder.transfer(fchannel, pos, 1); Assert.assertEquals(1, bytesRead); Assert.assertFalse(decoder.isCompleted()); Assert.assertEquals(0, metrics.getBytesTransferred()); pos += bytesRead; bytesRead = decoder.transfer(fchannel, pos, 2); Assert.assertEquals(2, bytesRead); Assert.assertFalse(decoder.isCompleted()); Assert.assertEquals(0, metrics.getBytesTransferred()); pos += bytesRead; bytesRead = decoder.transfer(fchannel, pos, 17); Assert.assertEquals(16, bytesRead); Assert.assertFalse(decoder.isCompleted()); Assert.assertEquals(0, metrics.getBytesTransferred()); pos += bytesRead; bytesRead = decoder.transfer(fchannel, pos, 1); Assert.assertEquals(1, bytesRead); Assert.assertFalse(decoder.isCompleted()); Assert.assertEquals(1, metrics.getBytesTransferred()); pos += bytesRead; bytesRead = decoder.transfer(fchannel, pos, 2); Assert.assertEquals(2, bytesRead); Assert.assertFalse(decoder.isCompleted()); Assert.assertEquals(3, metrics.getBytesTransferred()); pos += bytesRead; bytesRead = decoder.transfer(fchannel, pos, 15); Assert.assertEquals(14, bytesRead); Assert.assertTrue(decoder.isCompleted()); Assert.assertEquals(17, metrics.getBytesTransferred()); pos += bytesRead; bytesRead = decoder.transfer(fchannel, pos, 1); Assert.assertEquals(-1, bytesRead); Assert.assertTrue(decoder.isCompleted()); Assert.assertEquals(17, metrics.getBytesTransferred()); } finally { testfile.close(); } Assert.assertEquals("stuff; more stuff; a lot more stuff!", CodecTestUtils.readFromFile(this.tmpfile)); }
@Test public void testWriteBeyondFileSize() throws Exception { final ReadableByteChannel channel = new ReadableByteChannelMock( new String[] {"a"}, StandardCharsets.US_ASCII); final SessionInputBuffer inbuf = new SessionInputBufferImpl(1024, 256, 0, StandardCharsets.US_ASCII); final BasicHttpTransportMetrics metrics = new BasicHttpTransportMetrics(); final LengthDelimitedDecoder decoder = new LengthDelimitedDecoder( channel, inbuf, metrics, 1); createTempFile(); final RandomAccessFile testfile = new RandomAccessFile(this.tmpfile, "rw"); try { final FileChannel fchannel = testfile.getChannel(); Assert.assertEquals(0, testfile.length()); try { decoder.transfer(fchannel, 5, 10); Assert.fail("IOException should have been thrown"); } catch(final IOException expected) { } } finally { testfile.close(); } }
@Test public void testCodingBeyondContentLimitFile() throws Exception { final ReadableByteChannel channel = new ReadableByteChannelMock( new String[] { "stuff;", "more stuff; and a lot more stuff"}, StandardCharsets.US_ASCII); final SessionInputBuffer inbuf = new SessionInputBufferImpl(1024, 256, 0, StandardCharsets.US_ASCII); final BasicHttpTransportMetrics metrics = new BasicHttpTransportMetrics(); final LengthDelimitedDecoder decoder = new LengthDelimitedDecoder( channel, inbuf, metrics, 16); createTempFile(); final RandomAccessFile testfile = new RandomAccessFile(this.tmpfile, "rw"); try { final FileChannel fchannel = testfile.getChannel(); long bytesRead = decoder.transfer(fchannel, 0, 6); Assert.assertEquals(6, bytesRead); Assert.assertFalse(decoder.isCompleted()); Assert.assertEquals(6, metrics.getBytesTransferred()); bytesRead = decoder.transfer(fchannel,0 , 10); Assert.assertEquals(10, bytesRead); Assert.assertTrue(decoder.isCompleted()); Assert.assertEquals(16, metrics.getBytesTransferred()); bytesRead = decoder.transfer(fchannel, 0, 1); Assert.assertEquals(-1, bytesRead); Assert.assertTrue(decoder.isCompleted()); Assert.assertEquals(16, metrics.getBytesTransferred()); } finally { testfile.close(); } }
@Test(expected=ConnectionClosedException.class) public void testTruncatedContentWithFile() throws Exception { final ReadableByteChannel channel = new ReadableByteChannelMock( new String[] {"1234567890"}, StandardCharsets.US_ASCII); final SessionInputBuffer inbuf = new SessionInputBufferImpl(1024, 256, 0, StandardCharsets.US_ASCII); final BasicHttpTransportMetrics metrics = new BasicHttpTransportMetrics(); final LengthDelimitedDecoder decoder = new LengthDelimitedDecoder( channel, inbuf, metrics, 20); createTempFile(); final RandomAccessFile testfile = new RandomAccessFile(this.tmpfile, "rw"); try { final FileChannel fchannel = testfile.getChannel(); final long bytesRead = decoder.transfer(fchannel, 0, Integer.MAX_VALUE); Assert.assertEquals(10, bytesRead); decoder.transfer(fchannel, 0, Integer.MAX_VALUE); } finally { testfile.close(); } }