@Test public void testClose() throws Exception { monitor.start(); ReqId reqId1 = new ReqId(0, 1); ReqId reqId2 = new ReqId(1, 2); ReqId reqId3 = new ReqId(2, 3); ReqId reqId4 = new ReqId(3, 4); TransactionFuture future1 = monitor.register(reqId1, context, timeout); assertFalse(future1.isDone()); TransactionFuture future2 = monitor.register(reqId2, context, timeout); assertFalse(future2.isDone()); TransactionFuture future3 = monitor.register(reqId3, context, timeout); assertFalse(future3.isDone()); TransactionFuture future4 = monitor.register(reqId4, context, timeout); assertFalse(future4.isDone()); monitor.committed(reqId1); monitor.committed(reqId2); assertTrue(future1.isDone()); assertTrue(future1.get()); assertTrue(future1.isFlushed()); assertTrue(future2.isDone()); assertTrue(future2.get()); assertTrue(future2.isFlushed()); assertFalse(future3.isDone()); assertFalse(future3.isFlushed()); assertFalse(future4.isDone()); assertFalse(future4.isFlushed()); monitor.close(); assertTrue(future3.isDone()); assertTrue(future3.isFlushed()); try { future3.get(); fail(); } catch (ExecutionException ex) { assertTrue(ex.getCause() instanceof ClientClosedException); } catch (Throwable ex) { fail(); } assertTrue(future4.isDone()); assertTrue(future4.isFlushed()); try { future4.get(); fail(); } catch (ExecutionException ex) { assertTrue(ex.getCause() instanceof ClientClosedException); } catch (Throwable ex) { fail(); } }
public void close() { synchronized (this) { state = State.CLOSED; if (!registered.isEmpty()) { ClientClosedException exception = new ClientClosedException(); registered.values().forEach(future -> { future.completeExceptionally(exception); future.flushed(); }); registered.clear(); numRegistered = 0; } notifyAll(); } }
TransactionMonitor { public void close() { synchronized (this) { state = State.CLOSED; if (!registered.isEmpty()) { ClientClosedException exception = new ClientClosedException(); registered.values().forEach(future -> { future.completeExceptionally(exception); future.flushed(); }); registered.clear(); numRegistered = 0; } notifyAll(); } } }
TransactionMonitor { public void close() { synchronized (this) { state = State.CLOSED; if (!registered.isEmpty()) { ClientClosedException exception = new ClientClosedException(); registered.values().forEach(future -> { future.completeExceptionally(exception); future.flushed(); }); registered.clear(); numRegistered = 0; } notifyAll(); } } TransactionMonitor(int maxConcurrentTransactions); }
TransactionMonitor { public void close() { synchronized (this) { state = State.CLOSED; if (!registered.isEmpty()) { ClientClosedException exception = new ClientClosedException(); registered.values().forEach(future -> { future.completeExceptionally(exception); 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 close() { synchronized (this) { state = State.CLOSED; if (!registered.isEmpty()) { ClientClosedException exception = new ClientClosedException(); registered.values().forEach(future -> { future.completeExceptionally(exception); 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();  }