RewindableReadableByteChannel implements ReadableByteChannel { @Override public synchronized int read(ByteBuffer dst) throws IOException { if (directRead) { return baseChannel.read(dst); } int bytesToReadCount = dst.remaining(); if (bytesToReadCount == 0) { return 0; } if (buffer == null) { if (!canRewind) { directRead = true; return baseChannel.read(dst); } buffer = ByteBuffer.allocate(bytesToReadCount); int baseReadResult = baseChannel.read(buffer); buffer.flip(); if (baseReadResult > 0) { dst.put(buffer); } return baseReadResult; } if (buffer.remaining() >= bytesToReadCount) { int limit = buffer.limit(); buffer.limit(buffer.position() + bytesToReadCount); dst.put(buffer); buffer.limit(limit); if (!canRewind && !buffer.hasRemaining()) { buffer = null; directRead = true; } return bytesToReadCount; } int bytesFromBufferCount = buffer.remaining(); int stillToReadCount = bytesToReadCount - bytesFromBufferCount; int currentReadPos = buffer.position(); int contentLimit = buffer.limit(); setBufferLimit(contentLimit + stillToReadCount); buffer.position(contentLimit); int baseReadResult = baseChannel.read(buffer); buffer.flip(); buffer.position(currentReadPos); dst.put(buffer); if (bytesFromBufferCount == 0 && baseReadResult < 0) { return -1; } int bytesCount = buffer.position() - currentReadPos; if (!canRewind && !buffer.hasRemaining()) { buffer = null; directRead = true; } return bytesCount; } RewindableReadableByteChannel(ReadableByteChannel baseChannel); synchronized void disableRewinding(); synchronized void rewind(); @Override synchronized int read(ByteBuffer dst); @Override synchronized void close(); @Override synchronized boolean isOpen();  }
@Test public void testReadAfterEof() throws Exception { byte[] inputData = "The quick brown fox jumps over the lazy dog.".getBytes(UTF_8); ReadableByteChannel baseChannel = new ByteBufferChannel(inputData); RewindableReadableByteChannel rewindableChannel = new RewindableReadableByteChannel(baseChannel); ByteBuffer buffer1 = ByteBuffer.allocate(1000); assertEquals(inputData.length, rewindableChannel.read(buffer1)); ByteBuffer buffer2 = ByteBuffer.allocate(10); assertEquals(-1, rewindableChannel.read(buffer2)); assertArrayEquals(Arrays.copyOf(buffer2.array(), buffer2.position()), new byte[]{}); }
@Test public void testReadToEmptyBuffer() throws Exception { byte[] inputData = "The quick brown fox jumps over the lazy dog.".getBytes(UTF_8); ReadableByteChannel baseChannel = new ByteBufferChannel(inputData); RewindableReadableByteChannel rewindableChannel = new RewindableReadableByteChannel(baseChannel); ByteBuffer buffer = ByteBuffer.allocate(0); assertEquals(0, rewindableChannel.read(buffer)); }
@Test public void testNoData() throws Exception { byte[] inputData = "".getBytes(UTF_8); ReadableByteChannel baseChannel = new ByteBufferChannel(inputData); RewindableReadableByteChannel rewindableChannel = new RewindableReadableByteChannel(baseChannel); ByteBuffer buffer = ByteBuffer.allocate(10); assertEquals(-1, rewindableChannel.read(buffer)); }
@Test public void testReadEverything() throws Exception { byte[] inputData = "The quick brown fox jumps over the lazy dog.".getBytes(UTF_8); ReadableByteChannel baseChannel = new ByteBufferChannel(inputData); RewindableReadableByteChannel rewindableChannel = new RewindableReadableByteChannel(baseChannel); ByteBuffer buffer = ByteBuffer.allocate(1000); assertEquals(inputData.length, rewindableChannel.read(buffer)); assertArrayEquals(Arrays.copyOf(buffer.array(), buffer.position()), inputData); }
@Test public void testReadEverythingInChunks() throws Exception { byte[] inputData = "The quick brown fox jumps over the lazy dog.".getBytes(UTF_8); ReadableByteChannel baseChannel = new ByteBufferChannel(inputData, 20); RewindableReadableByteChannel rewindableChannel = new RewindableReadableByteChannel(baseChannel); ByteBuffer buffer = ByteBuffer.allocate(1000); assertEquals(20, rewindableChannel.read(buffer)); assertEquals(20, rewindableChannel.read(buffer)); assertEquals(inputData.length - 40, rewindableChannel.read(buffer)); assertEquals(-1, rewindableChannel.read(buffer)); assertArrayEquals(Arrays.copyOf(buffer.array(), inputData.length) , inputData); }
@Test public void testSmallChunksNoDataEveryOtherRead() throws Exception { byte[] inputData = "The quick brown fox jumps over the lazy dog.".getBytes(UTF_8); ReadableByteChannel baseChannel = new ByteBufferChannel(inputData, 10, true); RewindableReadableByteChannel rewindableChannel = new RewindableReadableByteChannel(baseChannel); ByteBuffer buffer1 = ByteBuffer.allocate(15); assertEquals(0, rewindableChannel.read(buffer1)); assertEquals(10, rewindableChannel.read(buffer1)); assertEquals(0, rewindableChannel.read(buffer1)); assertEquals(5, rewindableChannel.read(buffer1)); assertArrayEquals(buffer1.array(), Arrays.copyOf(inputData, 15)); }
@Test public void testReadToLimit() throws Exception { byte[] inputData = "The quick brown fox jumps over the lazy dog.".getBytes(UTF_8); ReadableByteChannel baseChannel = new ByteBufferChannel(inputData); RewindableReadableByteChannel rewindableChannel = new RewindableReadableByteChannel(baseChannel); ByteBuffer buffer = ByteBuffer.allocate(40); buffer.limit(10); assertEquals(10, rewindableChannel.read(buffer)); buffer.limit(30); assertEquals(20, rewindableChannel.read(buffer)); assertArrayEquals(Arrays.copyOf(buffer.array(), 30), Arrays.copyOf(inputData, 30)); }
@Test public void testReadLongInputByteByByte() throws Exception { int size = 400000; byte[] inputData = new byte[size]; for (int i = 0; i < size; i++) { inputData[i] = (byte) (i % 253); } ReadableByteChannel baseChannel = new ByteBufferChannel(inputData); RewindableReadableByteChannel rewindableChannel = new RewindableReadableByteChannel(baseChannel); ByteBuffer buffer = ByteBuffer.allocate(size); ByteBuffer byteBuffer = ByteBuffer.allocate(1); while (true) { int s = rewindableChannel.read(byteBuffer); byteBuffer.flip(); buffer.put(byteBuffer); if (s == -1) { break; } byteBuffer.clear(); } assertArrayEquals(Arrays.copyOf(buffer.array(), buffer.position()), inputData); }