Merge branch 'mogronalol-BAEL-343'

pom.xml

@@ -94,6 +94,7 @@
         <module>querydsl</module>
 
         <!-- <module>raml</module> -->
+        <module>reactor-core</module>
         <module>redis</module>
         <module>rest-assured</module>
         <module>rest-testing</module>

reactor-core/pom.xml

@@ -0,0 +1,59 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>org.baeldung</groupId>
+    <artifactId>reactor-core</artifactId>
+    <version>0.0.1-SNAPSHOT</version>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>1.8</source>
+                    <target>1.8</target>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-core</artifactId>
+            <version>${reactor-core.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>${junit.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <version>${assertj.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+            <version>${logback.version}</version>
+        </dependency>
+
+    </dependencies>
+
+    <properties>
+        <reactor-core.version>3.0.4.RELEASE</reactor-core.version>
+        <junit.version>4.12</junit.version>
+        <assertj.version>3.6.1</assertj.version>
+        <logback.version>1.1.3</logback.version>
+    </properties>
+
+</project>

reactor-core/src/test/java/com/baeldung/reactor/ReactorTest.java

@@ -0,0 +1,122 @@
+package com.baeldung.reactor;
+
+import org.junit.Test;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import reactor.core.publisher.ConnectableFlux;
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class ReactorTest {
+
+    @Test
+    public void givenFlux_whenSubscribing_thenStream() throws InterruptedException {
+
+        List<Integer> elements = new ArrayList<>();
+
+        Flux.just(1, 2, 3, 4)
+                .log()
+                .map(i -> {
+                    System.out.println(i + ":" + Thread.currentThread());
+                    return i * 2;
+                })
+                .subscribe(elements::add);
+
+        assertThat(elements).containsExactly(2, 4, 6, 8);
+    }
+
+    @Test
+    public void givenFlux_whenZipping_thenCombine() {
+        List<String> elements = new ArrayList<>();
+
+        Flux.just(1, 2, 3, 4)
+                .log()
+                .map(i -> i * 2)
+                .zipWith(Flux.range(0, Integer.MAX_VALUE).log(), (two, one) -> String.format("First Flux: %d, Second Flux: %d", one, two))
+                .subscribe(elements::add);
+
+        assertThat(elements).containsExactly(
+                "First Flux: 0, Second Flux: 2",
+                "First Flux: 1, Second Flux: 4",
+                "First Flux: 2, Second Flux: 6",
+                "First Flux: 3, Second Flux: 8");
+    }
+
+    @Test
+    public void givenFlux_whenApplyingBackPressure_thenPushElementsInBatches() throws InterruptedException {
+
+        List<Integer> elements = new ArrayList<>();
+
+        Flux.just(1, 2, 3, 4)
+                .log()
+                .map(i -> i * 2)
+                .onBackpressureBuffer()
+                .subscribe(new Subscriber<Integer>() {
+                    private Subscription s;
+                    int onNextAmount;
+
+                    @Override
+                    public void onSubscribe(final Subscription s) {
+                        this.s = s;
+                        s.request(2);
+                    }
+
+                    @Override
+                    public void onNext(final Integer integer) {
+                        elements.add(integer);
+                        onNextAmount++;
+                        if (onNextAmount % 2 == 0) {
+                            s.request(2);
+                        }
+                    }
+
+                    @Override
+                    public void onError(final Throwable t) {
+                    }
+
+                    @Override
+                    public void onComplete() {
+                        int ham = 2;
+                    }
+                });
+
+        assertThat(elements).containsExactly(2, 4, 6, 8);
+    }
+
+    @Test
+    public void givenFlux_whenInParallel_thenSubscribeInDifferentThreads() throws InterruptedException {
+        List<String> threadNames = new ArrayList<>();
+
+        Flux.just(1, 2, 3, 4)
+                .log()
+                .map(i -> Thread.currentThread().getName())
+                .subscribeOn(Schedulers.parallel())
+                .subscribe(threadNames::add);
+
+        Thread.sleep(1000);
+
+        assertThat(threadNames).containsExactly("parallel-1", "parallel-1", "parallel-1", "parallel-1");
+    }
+
+    @Test
+    public void givenConnectableFlux_thenShouldStream_onConnect() {
+
+        List<Integer> elements = new ArrayList<>();
+
+        final ConnectableFlux<Integer> publish = Flux.just(1, 2, 3, 4).publish();
+
+        publish.subscribe(elements::add);
+
+        assertThat(elements).isEmpty();
+
+        publish.connect();
+
+        assertThat(elements).containsExactly(1, 2, 3, 4);
+    }
+
+}