Add new flux examples and add new links to README (#34)

Author: marcosDLCS

README.adoc

@@ -1,6 +1,6 @@
 = Asynchronous & reactive programming in Java
 Marcos de la Calle Samaniego, @marcosDLCS <marcos.dlcs@gmail.com>
-v0.0.15, 2020-12-04
+v0.0.16, 2020-12-05
 :toc:
 
 ++++
@@ -250,6 +250,11 @@ Project Loom is to intended to explore, incubate and deliver Java VM features an
 |Mark Paluch
 |2020
 |🇬🇧 📋
+
+| https://www.javaadvent.com/2020/12/project-loom-and-structured-concurrency.html[Project Loom and structured concurrency, window=_blank]
+|Cay Horstmann
+|2020
+|🇬🇧 📋
 |===
 
 === ⚒️ Libraries and tools
@@ -327,15 +332,20 @@ Designed for both development and production time use
 |===
 |Name |Author |Platform |Lang
 
-| https://learning.oreilly.com/library/view/reactive-spring-boot/9780136836421/[Reactive Spring, 2nd Edition, window=_blank]
-|Josh Long
-|Oreilly
+| https://www.youtube.com/playlist?list=PLL8woMHwr36EDxjUoCzboZjedsnhLP1j4[Java concurrency and multithreading (playlist), window=_blank]
+|Jakob Jenkov
+|Youtube
 |🇬🇧
 
 | https://www.udemy.com/course/efficient-java-multithreading-with-executors/[Efficient Java Multithreading and Concurrency with Executors, window=_blank]
 |Arun Kumar
 |Udemy
 |🇬🇧
+
+| https://learning.oreilly.com/library/view/reactive-spring-boot/9780136836421/[Reactive Spring, 2nd Edition, window=_blank]
+|Josh Long
+|Oreilly
+|🇬🇧
 |===
 
 == 🚶‍♀️ Who to follow?

examples/00-reactor-operators/project-reactor-operators/src/test/java/es/codeurjc/arpj/FluxTest.java

@@ -5,6 +5,8 @@
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
 import reactor.util.function.Tuple2;
 
@@ -14,6 +16,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
+import java.util.function.Function;
 import java.util.stream.Stream;
 
 import static es.codeurjc.arpj.TestUtils.printSectionLine;
@@ -91,10 +94,10 @@ void flux_from_test() {
     }
 
     @Test
-    @DisplayName("Test 03: Flux transform")
-    void flux_transform() {
+    @DisplayName("Test 03: Flux map and cast")
+    void flux_map_cast() {
 
-        System.out.println("Test 03: Flux transform");
+        System.out.println("Test 03: Flux map and cast");
         printTestLine();
 
         final String[]      strings     = {"one", "two", "three", "four", "five"};
@@ -310,4 +313,102 @@ void flux_aggregate_reduce() {
                 .expectComplete()
                 .verify();
     }
+
+    @Test
+    @DisplayName("Test 08: Transform")
+    void flux_transform() {
+
+        System.out.println("Test 08: Transform");
+        printTestLine();
+
+        final Scheduler myParallel = Schedulers.newParallel("my-parallel", 2);
+
+        final Function<Flux<String>, Flux<Integer>> myTransformation =
+                flux -> flux.publishOn(myParallel).map(String::length);
+
+        final String[]      strings     = {"one", "two", "three", "four", "five"};
+        final Flux<Integer> mapToLength = Flux.fromArray(strings).transform(myTransformation).log();
+
+        StepVerifier
+                .create(mapToLength)
+                .expectNextCount(5)
+                .expectComplete()
+                .verify();
+    }
+
+    @Test
+    @DisplayName("Test 09: Concat map")
+    void flux_concat_map() {
+
+        System.out.println("Test 09: Concat map");
+        printTestLine();
+
+        final Scheduler myParallel = Schedulers.newParallel("my-parallel", 4);
+
+        final Flux<Integer> numbers = Flux.range(0, 10)
+                .window(2)
+                .concatMap(g -> g.flatMap(this::processing).subscribeOn(myParallel))
+                .log();
+
+        StepVerifier
+                .create(numbers)
+                .expectNextCount(10)
+                .expectComplete()
+                .verify();
+
+        printTestLine();
+
+        final Flux<Integer> numbers2 = Flux.range(0, 10)
+                .window(2)
+                .flatMap(g -> g.flatMap(this::processing).subscribeOn(myParallel))
+                .log();
+
+        StepVerifier
+                .create(numbers2)
+                .expectNextCount(10)
+                .expectComplete()
+                .verify();
+    }
+
+    private Mono<Integer> processing(final Integer input) {
+
+        final var random = new Random();
+
+        final var millis = random.nextInt(10) * 100L;
+
+        System.out.println("Sleeping [" + input + "] " + millis + " milliseconds... @ "
+                + Thread.currentThread().getName());
+
+        try {
+            Thread.sleep(millis);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+
+        return Mono.just(input);
+    }
+
+    @Test
+    @DisplayName("Test 10: Handle")
+    void flux_handle() {
+
+        System.out.println("Test 10: Handle");
+        printTestLine();
+
+        final Flux<Integer> numbers = Flux.range(0, 10)
+                .handle((n, s) -> {
+                    if (n % 2 == 0) {
+                        System.out.println("Even!!! -> " + n);
+                        s.next(n);
+                    } else {
+                        s.next(0);
+                    }
+                }).cast(Integer.class).log();
+
+        StepVerifier
+                .create(numbers)
+                .expectNextCount(10)
+                .expectComplete()
+                .verify();
+    }
 }