CoalescedPipe { public Object[] takeSnapshot() { _putLock.lock(); _takeLock.lock(); try { Object[] pendingElements = new Object[_pendingCount.get()]; ElementLink<E> currentElementLink = _headElementLink._nextElementLink; for (int i = 0; currentElementLink != null; i++) { pendingElements[i] = currentElementLink._element; currentElementLink = currentElementLink._nextElementLink; } return pendingElements; } finally { _putLock.unlock(); _takeLock.unlock(); } } CoalescedPipe();  CoalescedPipe(Comparator<E> comparator); long coalescedCount(); int pendingCount(); void put(E e); E take(); Object[] takeSnapshot();  }
@Test public void testTakeSnapshot() throws InterruptedException { CoalescedPipe<String> coalescedPipe = new CoalescedPipe<>(); Object[] snapShot = coalescedPipe.takeSnapshot(); Assert.assertEquals(Arrays.toString(snapShot), 0, snapShot.length); coalescedPipe.put("test1"); snapShot = coalescedPipe.takeSnapshot(); Assert.assertEquals(Arrays.toString(snapShot), 1, snapShot.length); Assert.assertEquals("test1", snapShot[0]); coalescedPipe.put("test2"); snapShot = coalescedPipe.takeSnapshot(); Assert.assertEquals(Arrays.toString(snapShot), 2, snapShot.length); Assert.assertEquals("test1", snapShot[0]); Assert.assertEquals("test2", snapShot[1]); }