HttpHelper { public int sendAll(Iterable<Observable<Integer>> batches, final int numMetrics, long timeoutMillis) { final AtomicBoolean err = new AtomicBoolean(false); final AtomicInteger updated = new AtomicInteger(0); LOGGER.debug("Got {} ms to send {} metrics", timeoutMillis, numMetrics); try { final CountDownLatch completed = new CountDownLatch(1); final Subscription s = Observable.mergeDelayError(Observable.from(batches)) .timeout(timeoutMillis, TimeUnit.MILLISECONDS) .subscribeOn(Schedulers.immediate()) .subscribe(updated::addAndGet, exc -> { logErr("onError caught", exc, updated.get(), numMetrics); err.set(true); completed.countDown(); }, completed::countDown); try { completed.await(timeoutMillis, TimeUnit.MILLISECONDS); } catch (InterruptedException interrupted) { err.set(true); s.unsubscribe(); LOGGER.warn("Timed out sending metrics. {}/{} sent", updated.get(), numMetrics); } } catch (Exception e) { err.set(true); logErr("Unexpected ", e, updated.get(), numMetrics); } if (updated.get() < numMetrics && !err.get()) { LOGGER.warn("No error caught, but only {}/{} sent.", updated.get(), numMetrics); } return updated.get(); } @Inject  HttpHelper(RxHttp rxHttp); RxHttp getRxHttp(); Observable<HttpClientResponse<ByteBuf>> postSmile(String uriStr, JsonPayload payload); int sendAll(Iterable<Observable<Integer>> batches,
                     final int numMetrics, long timeoutMillis); Response get(HttpClientRequest<ByteBuf> req, long timeout, TimeUnit timeUnit);  }
@Test public void testSendAll() throws Exception { List<Observable<Integer>> batches = new ArrayList<>(); int expectedSum = 0; for (int i = 1; i <= 5; ++i) { batches.add(Observable.just(i)); expectedSum += i; } HttpHelper httpHelper = new HttpHelper(null); int sent = httpHelper.sendAll(batches, expectedSum, 100L); assertEquals(sent, expectedSum); batches.add(Observable.<Integer>never()); int partial = httpHelper.sendAll(batches, expectedSum, 100L); assertEquals(partial, expectedSum); }
@Test public void testSendAllSlow() throws Exception { Observable<Integer> interval = Observable.interval(400, TimeUnit.MILLISECONDS).map(l -> l.intValue() + 1); List<Observable<Integer>> batches = new ArrayList<>(); batches.add(interval); int expectedSum = 3; for (int i = 1; i <= 5; ++i) { batches.add(Observable.just(i)); expectedSum += i; } HttpHelper httpHelper = new HttpHelper(null); int partial = httpHelper.sendAll(batches, expectedSum, 1000L); assertEquals(partial, expectedSum); }