From 73026b65653fe38fded09d85212d108f5b43ab27 Mon Sep 17 00:00:00 2001
From: Andrew Morgan <andrew.morgan@opencredo.com>
Date: Wed, 22 Feb 2017 20:32:38 +0000
Subject: [PATCH] Added reactor core

---
 pom.xml                                       |   1 +
 reactor-core/pom.xml                          |  52 +++++++++
 .../com/baeldung/reactor/ReactorTest.java     | 107 ++++++++++++++++++
 3 files changed, 160 insertions(+)
 create mode 100644 reactor-core/pom.xml
 create mode 100644 reactor-core/src/test/java/com/baeldung/reactor/ReactorTest.java

diff --git a/pom.xml b/pom.xml
index 9d6c5931e3..014e4016c5 100644
--- a/pom.xml
+++ b/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>
diff --git a/reactor-core/pom.xml b/reactor-core/pom.xml
new file mode 100644
index 0000000000..017b59f42e
--- /dev/null
+++ b/reactor-core/pom.xml
@@ -0,0 +1,52 @@
+<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>3.0.4.RELEASE</version>
+        </dependency>
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.12</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <version>3.6.1</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+            <version>1.1.3</version>
+        </dependency>
+
+    </dependencies>
+
+</project>
diff --git a/reactor-core/src/test/java/com/baeldung/reactor/ReactorTest.java b/reactor-core/src/test/java/com/baeldung/reactor/ReactorTest.java
new file mode 100644
index 0000000000..a90346803e
--- /dev/null
+++ b/reactor-core/src/test/java/com/baeldung/reactor/ReactorTest.java
@@ -0,0 +1,107 @@
+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.publisher.FluxSink;
+import reactor.core.scheduler.Schedulers;
+
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Stream;
+
+import static java.time.Duration.ofSeconds;
+import static java.util.stream.Collectors.toList;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class ReactorTest {
+
+    @Test
+    public void givenFlux_whenSubscribing_shouldStream() throws InterruptedException {
+
+        List<Integer> elements = new ArrayList<>();
+
+        Flux.just(1, 2, 3, 4)
+                .log()
+                .map(i -> i * 2)
+                .subscribe(elements::add);
+
+        assertThat(elements).containsExactly(2, 4, 6, 8);
+    }
+
+    @Test
+    public void givenFlux_whenZipping_shouldCombine() {
+        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_shouldPushLessElements() 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_whenInParalle_shouldSubscribeInDifferentThreads() {
+        List<Integer> elements = new ArrayList<>();
+
+        Flux.just(1, 2, 3, 4)
+                .log()
+                .map(i -> i * 2)
+                .subscribeOn(Schedulers.parallel())
+                .subscribe(elements::add);
+
+        assertThat(elements).containsExactly(2, 4, 6, 8);
+    }
+
+}