@Test public void updateRouting_compression() throws Exception { final CountDownLatch firstUpdateLatch = new CountDownLatch(1); Mockito.when(delegate.updateRouting(ArgumentMatchers.any(), ArgumentMatchers.any())) .then((Answer<FluentFuture<? extends CommitInfo>>) invocationOnMock -> { firstUpdateLatch.countDown(); return FluentFutures.immediateNullFluentFuture(); }); final FluentFuture<? extends CommitInfo> outcome1 = reactor.updateRouting(route1, route2); firstUpdateLatch.await(1, TimeUnit.SECONDS); final FluentFuture<? extends CommitInfo> outcome2 = reactor.updateRouting(route2, route3); final FluentFuture<? extends CommitInfo> outcome3 = reactor.updateRouting(route3, route4); outcome3.get(1, TimeUnit.SECONDS); outcome1.get(1, TimeUnit.SECONDS); outcome2.get(5, TimeUnit.SECONDS); final InOrder inOrder = Mockito.inOrder(delegate); inOrder.verify(delegate).updateRouting(route1, route2); inOrder.verify(delegate).updateRouting(route2, route4); inOrder.verifyNoMoreInteractions(); }
@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();  }