@Test public void testClear() throws Exception { monitor.start(); final int numReqIds = 1000; State<Integer> totalTransactions = new State<>(0); Thread writer = new Thread(() -> { for (int i = 0; i < numReqIds; i++) { ReqId reqId = new ReqId(0, i); monitor.register(reqId, context, timeout); totalTransactions.set(i + 1); } }); Thread reader = new Thread(() -> { Random rand = new Random(); int index = -1; for (int i = 0; i < numReqIds; i++) { if (rand.nextInt(20) == 0) { Uninterruptibly.sleep(5); } while (true) { StateChangeFuture<Integer> future = totalTransactions.watch(); index = future.currentState; if (index > i) { break; } try { future.get(); } catch (Exception ex) { } } monitor.committed(new ReqId(0, i)); } }); Thread cleaner = new Thread(() -> { Random rand = new Random(); int index = -1; for (int i = 0; i < numReqIds; i++) { if (rand.nextInt(20) == 0) { Uninterruptibly.sleep(5); monitor.clear(); } while (true) { StateChangeFuture<Integer> future = totalTransactions.watch(); index = future.currentState; if (index > i) { break; } try { future.get(); } catch (Exception ex) { } } } }); cleaner.start(); writer.start(); reader.start(); Uninterruptibly.join(10000, writer, reader, cleaner); assertEquals(0, monitor.registeredCount()); }
public void clear() { synchronized (this) { registered.values().forEach(future -> { future.complete(false); future.flushed(); }); registered.clear(); numRegistered = 0; notifyAll(); } }
TransactionMonitor { public void clear() { synchronized (this) { registered.values().forEach(future -> { future.complete(false); future.flushed(); }); registered.clear(); numRegistered = 0; notifyAll(); } } }
TransactionMonitor { public void clear() { synchronized (this) { registered.values().forEach(future -> { future.complete(false); future.flushed(); }); registered.clear(); numRegistered = 0; notifyAll(); } } TransactionMonitor(int maxConcurrentTransactions); }
TransactionMonitor { public void clear() { synchronized (this) { registered.values().forEach(future -> { future.complete(false); future.flushed(); }); registered.clear(); numRegistered = 0; notifyAll(); } } TransactionMonitor(int maxConcurrentTransactions); void close(); boolean start(); boolean stop(); boolean isStopped(); void clear(); int maxCapacity(); TransactionFuture register(ReqId reqId, TransactionContext context, long registrationTimeout); TransactionContext getTransactionContext(ReqId reqId); TransactionContext committed(ReqId reqId); void abort(ReqId reqId); void flush(ReqId reqId); TransactionFuture lastEnqueued(); long lastEnqueuedTime(); int registeredCount(); }
TransactionMonitor { public void clear() { synchronized (this) { registered.values().forEach(future -> { future.complete(false); future.flushed(); }); registered.clear(); numRegistered = 0; notifyAll(); } } TransactionMonitor(int maxConcurrentTransactions); void close(); boolean start(); boolean stop(); boolean isStopped(); void clear(); int maxCapacity(); TransactionFuture register(ReqId reqId, TransactionContext context, long registrationTimeout); TransactionContext getTransactionContext(ReqId reqId); TransactionContext committed(ReqId reqId); void abort(ReqId reqId); void flush(ReqId reqId); TransactionFuture lastEnqueued(); long lastEnqueuedTime(); int registeredCount();  }