CounterWait { public long waitFor(long value) throws InterruptedException, TimeoutException { if (value <= currentId.get()) { LOGGER.debug("Value {} reached", value); return currentId.get(); } ValueEvent eid = new ValueEvent(value); waiters.put(eid); if (value <= currentId.get()) { LOGGER.debug("Value {} reached", value); return currentId.get(); } LOGGER.debug("Blocked, waiting for value {}", value); eid.waitFor(); return currentId.get(); } CounterWait();  CounterWait(long waitTimeoutSec);  CounterWait(long waitTimeout, TimeUnit waitTimeUnit); synchronized void update(long newValue); synchronized void reset(long newValue); long waitFor(long value); @Override String toString(); int waitersCount();  }
@Test public void testWaitFor() throws Exception { int nthreads = 20; ExecutorService executor = Executors.newFixedThreadPool(nthreads); final CounterWait waiter = new CounterWait(); assertEquals(0, waiter.waitFor(0)); for (int i = 0; i < nthreads; i++) { int finalI = i + 2; final int val = finalI / 2; executor.execute(new Runnable() { public void run() { long r = 0; try { r = waiter.waitFor(val); } catch (InterruptedException | TimeoutException e) { e.printStackTrace(); } outSyncQueue.add(r); } } ); } while(waiter.waitersCount() < nthreads) { sleep(20); } assertTrue(outSyncQueue.isEmpty()); for (int i = 0; i < (nthreads / 2); i++) { waiter.update(i + 1); long r = outSyncQueue.takeFirst(); assertEquals(r, i + 1); r = outSyncQueue.takeFirst(); assertEquals(r, i + 1); assertTrue(outSyncQueue.isEmpty()); } executor.shutdown(); }
@Test(expected = TimeoutException.class) public void testWaitForWithTimeout() throws Exception { CounterWait waiter = new CounterWait(1, TimeUnit.MILLISECONDS); waiter.waitFor(1); }