CoalescedPipe { public void put(E e) throws InterruptedException { if (e == null) { throw new NullPointerException(); } int pendingElements = -1; _putLock.lockInterruptibly(); try { if (_coalesceElement(e)) { return; } _lastElementLink._nextElementLink = new ElementLink<>(e); _lastElementLink = _lastElementLink._nextElementLink; pendingElements = _pendingCount.getAndIncrement(); } finally { _putLock.unlock(); } if (pendingElements == 0) { _takeLock.lock(); try { _notEmptyCondition.signal(); } finally { _takeLock.unlock(); } } } CoalescedPipe();  CoalescedPipe(Comparator<E> comparator); long coalescedCount(); int pendingCount(); void put(E e); E take(); Object[] takeSnapshot();  }
@Test public void testPut() throws InterruptedException { CoalescedPipe<String> coalescedPipe = new CoalescedPipe<>(); try { coalescedPipe.put(null); Assert.fail(); } catch (NullPointerException nullPointerException) { } coalescedPipe.put("test1"); Assert.assertEquals(1, coalescedPipe.pendingCount()); Assert.assertEquals(0, coalescedPipe.coalescedCount()); coalescedPipe.put("test2"); Assert.assertEquals(2, coalescedPipe.pendingCount()); Assert.assertEquals(0, coalescedPipe.coalescedCount()); coalescedPipe.put("test1"); Assert.assertEquals(2, coalescedPipe.pendingCount()); Assert.assertEquals(1, coalescedPipe.coalescedCount()); coalescedPipe.put("test2"); Assert.assertEquals(2, coalescedPipe.pendingCount()); Assert.assertEquals(2, coalescedPipe.coalescedCount()); coalescedPipe = new CoalescedPipe<String>( new Comparator<String>() { @Override public int compare(String o1, String o2) { return o1.length() - o2.length(); } }); try { coalescedPipe.put(null); Assert.fail(); } catch (NullPointerException nullPointerException) { } coalescedPipe.put("a"); Assert.assertEquals(1, coalescedPipe.pendingCount()); Assert.assertEquals(0, coalescedPipe.coalescedCount()); coalescedPipe.put("ab"); Assert.assertEquals(2, coalescedPipe.pendingCount()); Assert.assertEquals(0, coalescedPipe.coalescedCount()); coalescedPipe.put("c"); Assert.assertEquals(2, coalescedPipe.pendingCount()); Assert.assertEquals(1, coalescedPipe.coalescedCount()); coalescedPipe.put("cd"); Assert.assertEquals(2, coalescedPipe.pendingCount()); Assert.assertEquals(2, coalescedPipe.coalescedCount()); }