CoalescingSupplier implements Supplier<T> { @Override public T get() { Round present = round; if (present.isFirstToArrive()) { present.execute(); return present.getResult(); } Round next = present.awaitDone(); if (next.isFirstToArrive()) { next.execute(); } return next.getResult(); } CoalescingSupplier(Supplier<T> delegate); @Override T get(); ListenableFuture<T> getAsync();  }
@Test public void delegatesToDelegate() { assertThat(supplier.get()).isEqualTo(DEFAULT_VALUE); verify(delegate).get(); verifyNoMoreInteractions(delegate); }
@Test public void batchesConcurrentRequests() throws InterruptedException { freezableDelegate.freeze(); AsyncTasks initialTask = getConcurrently(1); AsyncTasks batch = getConcurrently(5); freezableDelegate.unfreeze(); batch.await(); verify(delegate, atLeast(2)).get(); verify(delegate, atMost(5)).get(); }
@Test public void doesNotBatchSerialRequests() { supplier.get(); supplier.get(); supplier.get(); verify(delegate, times(3)).get(); }
@Test public void requestsDoNotRecieveOldResults() { assertThat(supplier.get()).isEqualTo(DEFAULT_VALUE); int value2 = 2; when(delegate.get()).thenReturn(value2); assertThat(supplier.get()).isEqualTo(value2); }
@Test public void exceptionsArePropagated() { RuntimeException expected = new RuntimeException("foo"); when(delegate.get()).thenThrow(expected); assertThatThrownBy(() -> supplier.get()).hasMessage(expected.getMessage()); }
@Test public void exceptionsArePropagatedForCoalescedCalls() { RuntimeException expected = new RuntimeException("foo"); when(delegate.get()).thenThrow(expected); freezableDelegate.freeze(); AsyncTasks tasks = getConcurrently(5); freezableDelegate.unfreeze(); tasks.assertAllFailed(expected); }