@SuppressWarnings("unchecked") @Test public void testExecuteAllReturnsValueInFuture() throws Exception { ServiceEndPoint FOO = mock(ServiceEndPoint.class); ServiceEndPoint BAR = mock(ServiceEndPoint.class); ServiceEndPoint BAZ = mock(ServiceEndPoint.class); when(_mockPool.getAllEndPoints()).thenReturn(Lists.newArrayList(FOO, BAR, BAZ)); when(_mockPool.executeOnEndPoint(same(FOO), any(ServiceCallback.class))).thenReturn("FOO"); when(_mockPool.executeOnEndPoint(same(BAR), any(ServiceCallback.class))).thenReturn("BAR"); when(_mockPool.executeOnEndPoint(same(BAZ), any(ServiceCallback.class))).thenReturn("BAZ"); AsyncServicePool<Service> pool = newAsyncPool(MoreExecutors.newDirectExecutorService()); Collection<Future<String>> futures = pool.executeOnAll(NEVER_RETRY, mock(ServiceCallback.class)); assertEquals(3, futures.size()); Set<String> results = Sets.newHashSet(); for (Future<String> future : futures) { results.add(future.get()); } assertEquals(Sets.newHashSet("FOO", "BAR", "BAZ"), results); }
@Override public <R> Collection<Future<R>> executeOnAll(RetryPolicy retry, ServiceCallback<S, R> callback) { return executeOn(ALL_END_POINTS, retry, callback); }
AsyncServicePool implements com.bazaarvoice.ostrich.AsyncServicePool<S> { @Override public <R> Collection<Future<R>> executeOnAll(RetryPolicy retry, ServiceCallback<S, R> callback) { return executeOn(ALL_END_POINTS, retry, callback); } }
AsyncServicePool implements com.bazaarvoice.ostrich.AsyncServicePool<S> { @Override public <R> Collection<Future<R>> executeOnAll(RetryPolicy retry, ServiceCallback<S, R> callback) { return executeOn(ALL_END_POINTS, retry, callback); } AsyncServicePool(Ticker ticker, ServicePool<S> pool, boolean shutdownPoolOnClose,
                            ExecutorService executor, boolean shutdownExecutorOnClose, MetricRegistry metrics); }
AsyncServicePool implements com.bazaarvoice.ostrich.AsyncServicePool<S> { @Override public <R> Collection<Future<R>> executeOnAll(RetryPolicy retry, ServiceCallback<S, R> callback) { return executeOn(ALL_END_POINTS, retry, callback); } AsyncServicePool(Ticker ticker, ServicePool<S> pool, boolean shutdownPoolOnClose,
                            ExecutorService executor, boolean shutdownExecutorOnClose, MetricRegistry metrics); @Override void close(); @Override Future<R> execute(final RetryPolicy retryPolicy, final ServiceCallback<S, R> callback); @Override Future<R> execute(final PartitionContext partitionContext, final RetryPolicy retryPolicy,
                                 final ServiceCallback<S, R> callback); @Override Collection<Future<R>> executeOnAll(RetryPolicy retry, ServiceCallback<S, R> callback); @Override Collection<Future<R>> executeOn(ServiceEndPointPredicate predicate, final RetryPolicy retry,
                                               final ServiceCallback<S, R> callback); @Override int getNumValidEndPoints(); @Override int getNumBadEndPoints(); }
AsyncServicePool implements com.bazaarvoice.ostrich.AsyncServicePool<S> { @Override public <R> Collection<Future<R>> executeOnAll(RetryPolicy retry, ServiceCallback<S, R> callback) { return executeOn(ALL_END_POINTS, retry, callback); } AsyncServicePool(Ticker ticker, ServicePool<S> pool, boolean shutdownPoolOnClose,
                            ExecutorService executor, boolean shutdownExecutorOnClose, MetricRegistry metrics); @Override void close(); @Override Future<R> execute(final RetryPolicy retryPolicy, final ServiceCallback<S, R> callback); @Override Future<R> execute(final PartitionContext partitionContext, final RetryPolicy retryPolicy,
                                 final ServiceCallback<S, R> callback); @Override Collection<Future<R>> executeOnAll(RetryPolicy retry, ServiceCallback<S, R> callback); @Override Collection<Future<R>> executeOn(ServiceEndPointPredicate predicate, final RetryPolicy retry,
                                               final ServiceCallback<S, R> callback); @Override int getNumValidEndPoints(); @Override int getNumBadEndPoints();  }