BufferedElementCountingOutputStream extends OutputStream { public void finish() throws IOException { if (finished) { return; } flush(); VarInt.encode(terminatorValue, os); if (!BUFFER_POOL.offer(buffer)) { } finished = true; } BufferedElementCountingOutputStream(OutputStream os);  BufferedElementCountingOutputStream(OutputStream os, long terminatorValue);   BufferedElementCountingOutputStream(OutputStream os, int bufferSize, long terminatorValue); void finish(); void markElementStart(); @Override void write(int b); @Override void write(byte[] b, int off, int len); @Override void flush(); @Override void close(); static final int DEFAULT_BUFFER_SIZE; }
@Test public void testFinishingWhenFinishedIsNoOp() throws Exception { BufferedElementCountingOutputStream os = testValues(toBytes("a")); os.finish(); os.finish(); os.finish(); }
@Test public void testBuffersAreTakenAndReturned() throws Exception { BUFFER_POOL.clear(); BUFFER_POOL.offer(ByteBuffer.allocate(256)); ByteArrayOutputStream baos = new ByteArrayOutputStream(); BufferedElementCountingOutputStream os = createAndWriteValues(toBytes("abcdefghij"), baos); assertEquals(0, BUFFER_POOL.size()); os.finish(); assertEquals(1, BUFFER_POOL.size()); }
@Test public void testBehaviorWhenBufferPoolFull() throws Exception { ByteArrayOutputStream baos = new ByteArrayOutputStream(); while (BUFFER_POOL.remainingCapacity() > 0) { BUFFER_POOL.offer(ByteBuffer.allocate(256)); } BufferedElementCountingOutputStream os = createAndWriteValues(toBytes("abcdefghij"), baos); os.finish(); assertEquals(0, BUFFER_POOL.remainingCapacity()); }
@Test public void testBehaviorWhenBufferPoolEmpty() throws Exception { BUFFER_POOL.clear(); ByteArrayOutputStream baos = new ByteArrayOutputStream(); BufferedElementCountingOutputStream os = createAndWriteValues(toBytes("abcdefghij"), baos); assertEquals(0, BUFFER_POOL.size()); os.finish(); assertEquals(1, BUFFER_POOL.size()); }