WeatherWebApi implements AutoCloseable { public CompletableFuture<Stream<WeatherInfo>> pastWeather(double lat, double log, LocalDate from, LocalDate to) { String path = HOST + String.format(PATH_PAST_WEATHER, lat, log, from, to, WEATHER_KEY); System.out.println(path); return ahc .prepareGet(path) .execute() .toCompletableFuture() .thenApply(Response::getResponseBody) .thenApply(body -> { PastWeatherDto dto = gson.fromJson(body, PastWeatherDto.class); return fromDtoToWeatherInfo(dto); }); }  CompletableFuture<Stream<WeatherInfo>> pastWeather(double lat, double log, LocalDate from, LocalDate to); @Override void close();  }
@Test public void pastWeather() throws IOException { CompletableFuture<Stream<WeatherInfo>> infos = new WeatherWebApi() .pastWeather(37.017, -7.933, parse("2019-02-01"), parse("2019-03-28")); WeatherInfo wi = infos.join().findFirst().get(); assertEquals(parse("2019-02-01"), wi.getDate()); assertEquals(14, wi.getTempC()); assertEquals(13.8, wi.getPrecipMM()); assertEquals("Patchy rain possible", wi.getDesc()); }
@Test public void pastWeatherToObservable() throws IOException { CompletableFuture<Stream<WeatherInfo>> infos = new WeatherWebApi() .pastWeather(37.017, -7.933, parse("2019-02-01"), parse("2019-03-28")); Observable<WeatherInfo> wis = Observable .fromFuture(infos) .flatMap(strm -> fromIterable(strm::iterator)); WeatherInfo wi = wis.firstElement().blockingGet(); assertEquals(parse("2019-02-01"), wi.getDate()); assertEquals(14, wi.getTempC()); assertEquals(13.8, wi.getPrecipMM()); assertEquals("Patchy rain possible", wi.getDesc()); }