@Test public void wipeRouting_simple() throws Exception { Mockito.when(delegate.wipeRouting()).thenReturn(FluentFutures.immediateNullFluentFuture()); reactor.wipeRouting().get(1, TimeUnit.SECONDS); Mockito.verify(delegate).wipeRouting(); }
@Override public FluentFuture<? extends CommitInfo> wipeRouting() { FluentFuture<? extends CommitInfo> futureResult; try { queueGuard.acquire(); if (compressionQueue.isEmpty()) { compressionQueue.add(new MutablePair<>(null, WIPE_ROUTING_MARK)); futureResult = FluentFuture.from(servicePool.submit(updateRouteTask)); } else { compressionQueue.peek().setRight(WIPE_ROUTING_MARK); futureResult = COMPRESSED_FUTURE_RESULT; LOG.trace("route update request got state compressed - firing immediate future"); } queueGuard.release(); } catch (Exception e) { LOG.warn("failed to get lock upon compression queue: {}", e.getMessage()); futureResult = FluentFutures.immediateFailedFluentFuture(e); } return futureResult; }
RouteReactorZipImpl implements RouteReactor { @Override public FluentFuture<? extends CommitInfo> wipeRouting() { FluentFuture<? extends CommitInfo> futureResult; try { queueGuard.acquire(); if (compressionQueue.isEmpty()) { compressionQueue.add(new MutablePair<>(null, WIPE_ROUTING_MARK)); futureResult = FluentFuture.from(servicePool.submit(updateRouteTask)); } else { compressionQueue.peek().setRight(WIPE_ROUTING_MARK); futureResult = COMPRESSED_FUTURE_RESULT; LOG.trace("route update request got state compressed - firing immediate future"); } queueGuard.release(); } catch (Exception e) { LOG.warn("failed to get lock upon compression queue: {}", e.getMessage()); futureResult = FluentFutures.immediateFailedFluentFuture(e); } return futureResult; } }
RouteReactorZipImpl implements RouteReactor { @Override public FluentFuture<? extends CommitInfo> wipeRouting() { FluentFuture<? extends CommitInfo> futureResult; try { queueGuard.acquire(); if (compressionQueue.isEmpty()) { compressionQueue.add(new MutablePair<>(null, WIPE_ROUTING_MARK)); futureResult = FluentFuture.from(servicePool.submit(updateRouteTask)); } else { compressionQueue.peek().setRight(WIPE_ROUTING_MARK); futureResult = COMPRESSED_FUTURE_RESULT; LOG.trace("route update request got state compressed - firing immediate future"); } queueGuard.release(); } catch (Exception e) { LOG.warn("failed to get lock upon compression queue: {}", e.getMessage()); futureResult = FluentFutures.immediateFailedFluentFuture(e); } return futureResult; } RouteReactorZipImpl(RouteReactor delegate); }
RouteReactorZipImpl implements RouteReactor { @Override public FluentFuture<? extends CommitInfo> wipeRouting() { FluentFuture<? extends CommitInfo> futureResult; try { queueGuard.acquire(); if (compressionQueue.isEmpty()) { compressionQueue.add(new MutablePair<>(null, WIPE_ROUTING_MARK)); futureResult = FluentFuture.from(servicePool.submit(updateRouteTask)); } else { compressionQueue.peek().setRight(WIPE_ROUTING_MARK); futureResult = COMPRESSED_FUTURE_RESULT; LOG.trace("route update request got state compressed - firing immediate future"); } queueGuard.release(); } catch (Exception e) { LOG.warn("failed to get lock upon compression queue: {}", e.getMessage()); futureResult = FluentFutures.immediateFailedFluentFuture(e); } return futureResult; } RouteReactorZipImpl(RouteReactor delegate); @Override FluentFuture<? extends CommitInfo> updateRouting(@Nullable final SxpClusterRoute oldRoute,
                                                            @Nullable final SxpClusterRoute newRoute); @Override FluentFuture<? extends CommitInfo> wipeRouting(); }
RouteReactorZipImpl implements RouteReactor { @Override public FluentFuture<? extends CommitInfo> wipeRouting() { FluentFuture<? extends CommitInfo> futureResult; try { queueGuard.acquire(); if (compressionQueue.isEmpty()) { compressionQueue.add(new MutablePair<>(null, WIPE_ROUTING_MARK)); futureResult = FluentFuture.from(servicePool.submit(updateRouteTask)); } else { compressionQueue.peek().setRight(WIPE_ROUTING_MARK); futureResult = COMPRESSED_FUTURE_RESULT; LOG.trace("route update request got state compressed - firing immediate future"); } queueGuard.release(); } catch (Exception e) { LOG.warn("failed to get lock upon compression queue: {}", e.getMessage()); futureResult = FluentFutures.immediateFailedFluentFuture(e); } return futureResult; } RouteReactorZipImpl(RouteReactor delegate); @Override FluentFuture<? extends CommitInfo> updateRouting(@Nullable final SxpClusterRoute oldRoute,
                                                            @Nullable final SxpClusterRoute newRoute); @Override FluentFuture<? extends CommitInfo> wipeRouting();  }