Bael 776 vert.x and rxjava (#2529)

* Burlap & Hessian server added * Burlap & Hessian client work * Fixed main * Fixed formatting * Spring Remote example based on Burlap & Hessian runs in a JUnit test * Fixed main * Fixed formatting * Spring Remote example based on Burlap & Hessian runs in a JUnit test * Spring Remote example based on Burlap & Hessian runs in a JUnit test * Burlap & Hessian client work * Fixed main * Fixed main * Fixed formatting * Fixed formatting * Spring Remote example based on Burlap & Hessian runs in a JUnit test * Spring Remote example based on Burlap & Hessian runs in a JUnit test * Fixed POM * Vertx / RxJava example * BAEL-776 - Vertx and RxJava * BAEL-776 - Vertx and RxJava
2017-08-30 11:20:56 +02:00
parent 7df2635822
commit 4c7f165c7e
9 changed files with 319 additions and 0 deletions
@@ -0,0 +1,94 @@
+package com.baeldung;
+
+import io.reactivex.Flowable;
+import io.vertx.reactivex.core.Vertx;
+import io.vertx.reactivex.core.buffer.Buffer;
+import io.vertx.reactivex.core.file.FileSystem;
+import io.vertx.reactivex.core.http.HttpClient;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.ZonedDateTime;
+
+import static com.baeldung.MetaWeatherClient.getDataByPlaceId;
+import static com.baeldung.MetaWeatherClient.searchByCityName;
+
+public class VertxWithRxJavaTest {
+
+    private Vertx vertx;
+    private HttpClient httpClient;
+    private FileSystem fileSystem;
+    static Logger log = LoggerFactory.getLogger(VertxWithRxJavaTest.class);
+
+    @Before public void setUp() {
+        vertx = io.vertx.reactivex.core.Vertx.vertx();
+        httpClient = vertx.createHttpClient();
+        fileSystem = vertx.fileSystem();
+    }
+
+    @After public void tearDown() {
+        vertx.close();
+    }
+
+    @Test public void lightLengthTest() throws InterruptedException {
+
+        // read the file that contains one city name per line
+        fileSystem
+                .rxReadFile("cities.txt").toFlowable()
+                .doOnNext(buffer -> log.info("File buffer ---\n{}\n---", buffer))
+
+                // split file content in lines to obtain one city per line
+                .flatMap(buffer -> Flowable.fromArray(buffer.toString().split("\\r?\\n")))
+                .doOnNext(city -> log.info("City from file: '{}'", city))
+
+                // discard cities that are commented out with a leading '#'
+                .filter(city -> !city.startsWith("#"))
+                .doOnNext(city -> log.info("City that survived filtering: '{}'", city))
+
+                // for each city sends a request to obtain its 'woeid'
+                // and collect the buffer from the answer
+                .flatMap(city -> searchByCityName(httpClient, city) )
+                .flatMap(response -> response.toFlowable())
+                .doOnNext(buffer -> log.info("JSON of city detail: '{}'", buffer))
+
+                // get the woeid of each city
+                .map(cityBuffer -> cityBuffer
+                        .toJsonArray()
+                        .getJsonObject(0)
+                        .getLong("woeid"))
+
+                // use the id to ask for data
+                .flatMap(cityId -> getDataByPlaceId(httpClient, cityId))
+                .flatMap(response -> response
+                        .toObservable()
+                        .reduce(
+                                Buffer.buffer(),
+                                (total, newBuf) -> total.appendBuffer( newBuf )).toFlowable() )
+
+                // get the JSON object out of the response
+                .doOnNext(buffer -> log.info("JSON of place detail: '{}'", buffer))
+                .map(buffer -> buffer.toJsonObject())
+
+                // map the JSON in a POJO
+                .map(json -> {
+                    ZonedDateTime sunRise = ZonedDateTime.parse(json.getString("sun_rise"));
+                    ZonedDateTime sunSet = ZonedDateTime.parse(json.getString("sun_set"));
+                    String cityName = json.getString("title");
+                    return new CityAndDayLength(
+                        cityName,sunSet.toEpochSecond() - sunRise.toEpochSecond());
+                })
+
+                // consume the events
+                .subscribe(
+                    cityAndDayLength -> System.out.println(cityAndDayLength),
+                    ex -> ex.printStackTrace());
+
+        // enough to give time to complete the execution
+        Thread.sleep(20000);
+
+    }
+
+}