@Test public void updateRouting_simple() throws Exception { Mockito.when(delegate.updateRouting(route1, route2)).thenReturn(FluentFutures.immediateNullFluentFuture()); reactor.updateRouting(route1, route2).get(1, TimeUnit.SECONDS); Mockito.verify(delegate).updateRouting(route1, route2); }
@Override public FluentFuture<? extends CommitInfo> updateRouting(@Nullable final SxpClusterRoute oldRoute, @Nullable final SxpClusterRoute newRoute) { FluentFuture<? extends CommitInfo> futureResult; try { queueGuard.acquire(); if (compressionQueue.isEmpty()) { compressionQueue.add(new MutablePair<>(oldRoute, newRoute)); futureResult = FluentFuture.from(servicePool.submit(updateRouteTask)); } else { compressionQueue.peek().setRight(newRoute); 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> updateRouting(@Nullable final SxpClusterRoute oldRoute, @Nullable final SxpClusterRoute newRoute) { FluentFuture<? extends CommitInfo> futureResult; try { queueGuard.acquire(); if (compressionQueue.isEmpty()) { compressionQueue.add(new MutablePair<>(oldRoute, newRoute)); futureResult = FluentFuture.from(servicePool.submit(updateRouteTask)); } else { compressionQueue.peek().setRight(newRoute); 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> updateRouting(@Nullable final SxpClusterRoute oldRoute, @Nullable final SxpClusterRoute newRoute) { FluentFuture<? extends CommitInfo> futureResult; try { queueGuard.acquire(); if (compressionQueue.isEmpty()) { compressionQueue.add(new MutablePair<>(oldRoute, newRoute)); futureResult = FluentFuture.from(servicePool.submit(updateRouteTask)); } else { compressionQueue.peek().setRight(newRoute); 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> updateRouting(@Nullable final SxpClusterRoute oldRoute, @Nullable final SxpClusterRoute newRoute) { FluentFuture<? extends CommitInfo> futureResult; try { queueGuard.acquire(); if (compressionQueue.isEmpty()) { compressionQueue.add(new MutablePair<>(oldRoute, newRoute)); futureResult = FluentFuture.from(servicePool.submit(updateRouteTask)); } else { compressionQueue.peek().setRight(newRoute); 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> updateRouting(@Nullable final SxpClusterRoute oldRoute, @Nullable final SxpClusterRoute newRoute) { FluentFuture<? extends CommitInfo> futureResult; try { queueGuard.acquire(); if (compressionQueue.isEmpty()) { compressionQueue.add(new MutablePair<>(oldRoute, newRoute)); futureResult = FluentFuture.from(servicePool.submit(updateRouteTask)); } else { compressionQueue.peek().setRight(newRoute); 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();  }