From 2b2b78cbf28072ed643c1b37cfd5929bada148b3 Mon Sep 17 00:00:00 2001
From: felipeazv <felipe.azv@gmail.com>
Date: Thu, 8 Feb 2018 10:24:21 +0100
Subject: [PATCH] [BAEL-1449]-Combining Publishers (Project Reactor)

---
 .../reactor/core/CombiningPublishersTest.java | 107 ++++++++++++++++++
 1 file changed, 107 insertions(+)
 create mode 100644 reactor-core/src/test/java/com/baeldung/reactor/core/CombiningPublishersTest.java

diff --git a/reactor-core/src/test/java/com/baeldung/reactor/core/CombiningPublishersTest.java b/reactor-core/src/test/java/com/baeldung/reactor/core/CombiningPublishersTest.java
new file mode 100644
index 0000000000..f33f911274
--- /dev/null
+++ b/reactor-core/src/test/java/com/baeldung/reactor/core/CombiningPublishersTest.java
@@ -0,0 +1,107 @@
+package com.baeldung.reactor.core;
+
+import java.time.Duration;
+
+import org.junit.Test;
+
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+public class CombiningPublishersTest {
+    
+    private static Integer min = 1;
+    private static Integer max = 5;
+
+    private static Flux<Integer> evenNumbers = Flux.range(min, max).filter(x -> x % 2 == 0);
+    private static Flux<Integer> oddNumbers = Flux.range(min, max).filter(x -> x % 2 > 0);
+    
+    
+    @Test
+    public void testMerge() {
+        Flux<Integer> fluxOfIntegers = Flux.merge(
+                evenNumbers, 
+                oddNumbers);
+        
+        StepVerifier.create(fluxOfIntegers)
+        .expectNext(2)
+        .expectNext(4)
+        .expectNext(1)
+        .expectNext(3)
+        .expectNext(5)
+        .expectComplete()
+        .verify();
+    }
+    
+    @Test
+    public void testMergeWithDelayedElements() {
+        Flux<Integer> fluxOfIntegers = Flux.merge(
+                evenNumbers.delayElements(Duration.ofMillis(500L)), 
+                oddNumbers.delayElements(Duration.ofMillis(300L)));
+        
+        StepVerifier.create(fluxOfIntegers)
+        .expectNext(1)
+        .expectNext(2)
+        .expectNext(3)
+        .expectNext(5)
+        .expectNext(4)
+        .expectComplete()
+        .verify();
+    }
+    
+    @Test
+    public void testConcat() {
+        Flux<Integer> fluxOfIntegers = Flux.concat(
+                evenNumbers.delayElements(Duration.ofMillis(500L)), 
+                oddNumbers.delayElements(Duration.ofMillis(300L)));
+        
+        StepVerifier.create(fluxOfIntegers)
+        .expectNext(2)
+        .expectNext(4)
+        .expectNext(1)
+        .expectNext(3)
+        .expectNext(5)
+        .expectComplete()
+        .verify();
+    }
+    
+    @Test
+    public void testConcatWith() {
+        Flux<Integer> fluxOfIntegers = evenNumbers
+                .concatWith(oddNumbers);
+        
+        StepVerifier.create(fluxOfIntegers)
+        .expectNext(2)
+        .expectNext(4)
+        .expectNext(1)
+        .expectNext(3)
+        .expectNext(5)
+        .expectComplete()
+        .verify();
+    }
+    
+    @Test
+    public void testCombineLatest() {
+        Flux<Integer> fluxOfIntegers = Flux.combineLatest(
+                evenNumbers, 
+                oddNumbers, 
+                (a, b) -> a + b);
+
+        StepVerifier.create(fluxOfIntegers)
+        .expectNext(5)
+        .expectNext(7)
+        .expectNext(9)
+        .expectComplete()
+        .verify();
+
+    }
+
+    @Test
+    public void testCombineLatest1() {
+            StepVerifier.create(Flux.combineLatest(obj -> (int) obj[1], evenNumbers, oddNumbers))
+            .expectNext(1)
+            .expectNext(3)
+            .expectNext(5)
+                        .verifyComplete();
+    }
+
+}