WeatherService implements AutoCloseable { public Observable<WeatherInfo> pastWeather(double lat, double log, LocalDate from, LocalDate to) { Stream<CompletableFuture<Stream<WeatherInfo>>> cfs = Stream .iterate(from, prev -> prev.withDayOfMonth(1).plusMonths(1)) .takeWhile(dt -> dt.isBefore(to)) .map(start -> { LocalDate end = min(start.withDayOfMonth(start.lengthOfMonth()), to); return webApi.pastWeather(lat, log, start, end); }); return Observable .fromIterable(cfs::iterator) .map(WeatherService::toObservable) .flatMap(Functions.identity()) .flatMap(strm -> Observable.fromIterable(strm::iterator)); }  Observable<WeatherInfo> pastWeather(double lat, double log, LocalDate from, LocalDate to); static Observable<T> toObservable(CompletableFuture<T> future); @Override void close();  }
@Test public void pastWeatherOnSeveralMonths() throws IOException { WeatherService weather = new WeatherService(); LocalDate[] from = {LocalDate.of(2018, 12, 7)}; LocalDate to = LocalDate.of(2019, 3, 19); Observable<WeatherInfo> infos = weather.pastWeather(37.017, -7.933, from[0], to); System.out.println(Thread.currentThread().getId()); CompletableFuture cf = new CompletableFuture(); infos.doOnNext(info -> { assertEquals(from[0], info.date); from[0] = from[0].plusDays(1); System.out.println(Thread.currentThread().getId()); }) .doOnComplete(() -> cf.complete(null)) .doOnError(err -> cf.completeExceptionally(err)) .subscribe(); cf.join(); assertEquals(from[0], to.plusDays(1)); }