@Test public void testFlush() throws Exception { monitor.start(); final int capacity = monitor.maxCapacity(); ArrayList<ReqId> reqIds = new ArrayList<>(); ArrayList<TransactionFuture> futures = new ArrayList<>(); for (int i = 0; i < capacity; i++) { reqIds.add(new ReqId(0, i)); futures.add(monitor.register(new ReqId(0, i), context, timeout)); } int aborted = capacity / 2; monitor.abort(reqIds.get(aborted)); int flushed = capacity / 2 - 1; monitor.flush(reqIds.get(flushed)); int remaining = 0; for (int i = 0; i < capacity; i++) { if (i <= flushed) { assertFalse(futures.get(i).get()); assertTrue(futures.get(i).isFlushed()); } else { if (i == aborted) { assertFalse(futures.get(i).get()); } else { assertFalse(futures.get(i).isDone()); remaining++; } assertFalse(futures.get(i).isFlushed()); } } assertEquals(remaining, monitor.registeredCount()); flushed = aborted; monitor.flush(reqIds.get(flushed)); remaining = 0; for (int i = 0; i < capacity; i++) { if (i <= flushed) { assertFalse(futures.get(i).get()); assertTrue(futures.get(i).isFlushed()); } else { assertFalse(futures.get(i).isDone()); assertFalse(futures.get(i).isFlushed()); remaining++; } } assertEquals(remaining, monitor.registeredCount()); flushed = capacity / 2 + 1; monitor.flush(reqIds.get(flushed)); remaining = 0; for (int i = 0; i < capacity; i++) { if (i <= flushed) { assertFalse(futures.get(i).get()); assertTrue(futures.get(i).isFlushed()); } else { assertFalse(futures.get(i).isDone()); assertFalse(futures.get(i).isFlushed()); remaining++; } } assertEquals(remaining, monitor.registeredCount()); flushed = capacity - 1; monitor.flush(reqIds.get(flushed)); for (int i = 0; i < capacity; i++) { assertFalse(futures.get(i).get()); assertTrue(futures.get(i).isFlushed()); } assertEquals(0, monitor.registeredCount()); }
public void flush(ReqId reqId) { synchronized (this) { if (state != State.CLOSED) { if (registered.containsKey(reqId)) { complete(reqId, false); notifyAll(); } } } }
TransactionMonitor { public void flush(ReqId reqId) { synchronized (this) { if (state != State.CLOSED) { if (registered.containsKey(reqId)) { complete(reqId, false); notifyAll(); } } } } }
TransactionMonitor { public void flush(ReqId reqId) { synchronized (this) { if (state != State.CLOSED) { if (registered.containsKey(reqId)) { complete(reqId, false); notifyAll(); } } } } TransactionMonitor(int maxConcurrentTransactions); }
TransactionMonitor { public void flush(ReqId reqId) { synchronized (this) { if (state != State.CLOSED) { if (registered.containsKey(reqId)) { complete(reqId, false); 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 flush(ReqId reqId) { synchronized (this) { if (state != State.CLOSED) { if (registered.containsKey(reqId)) { complete(reqId, false); 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();  }