code: 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();
    }
docs: Attempt to send all the batches totalling numMetrics in the allowed time.
  
  @return The total number of metrics sent.