reactive java

<dependency> <groupid>io.projectreactor</groupid> <artifactid>reactor-test</artifactid> <version>3.1.0.RELEASE</version> <scope>test</scope> </dependency> class ReactiveJavaTutorialTest { @Test public void testFlux() { //Create a Flux Flux <Integer> fluxToTest = Flux.just(1, 2, 3, 4, 5); // Create a test StepVerifier.create(fluxToTest) .expectNext(1) .expectNext(2) .expectNext(3) .expectNext(4) .expectNext(5) .expectComplete() // we expect Flux to complete after sending the number 5 which is the last element .verify(); }…

class ReactiveJavaTutorial { public static void main(String[] args) { Flux<String> cities = Flux.fromIterable( new ArrayList<>(Arrays.asList(“New York”, “London”, “Paris”, “Toronto”, “Rome”))); cities.log().subscribe(); } } Output: INFO 6832 — [ main] reactor.Flux.Iterable.1 : | onSubscribe([Synchronous Fuseable] FluxIterable.IterableSubscription) INFO 6832 — [ main] reactor.Flux.Iterable.1 : | request(unbounded) INFO 6832 — [ main] reactor.Flux.Iterable.1 : | onNext(New York) INFO…

class ReactiveJavaTutorial { public static void main(String[] args) { Flux<String> cities = Flux.just(“New York”, “London”, “Paris”, “Amsterdam”) .map(String::toUpperCase) .filter(cityName -> cityName.length() <= 8) .map(cityName -> cityName.concat(” City”)) .log(); cities.subscribe(); } Output: INFO 14040 — [main] reactor.Flux.MapFuseable.1 : | onSubscribe([Fuseable] FluxMapFuseable.MapFuseableSubscriber) INFO 14040 — [main] reactor.Flux.MapFuseable.1 : | request(unbounded) INFO 14040 — [main] reactor.Flux.MapFuseable.1 : |…

class ReactiveJavaTutorial { public static void main(String[] args) { Flux<String> cities = Flux.just(“New York”, “London”, “Paris”, “Amsterdam”) .map(String::toUpperCase) .filter(cityName -> cityName.length() <= 8) .map(cityName -> cityName.concat(” City”)) .log(); cities.subscribe(); } } INFO 14040 — [main] reactor.Flux.MapFuseable.1 : | onSubscribe([Fuseable] FluxMapFuseable.MapFuseableSubscriber) INFO 14040 — [main] reactor.Flux.MapFuseable.1 : | request(unbounded) INFO 14040 — [main] reactor.Flux.MapFuseable.1 : |…

class ReactiveJavaTutorial { public static void main(String[] args) throws InterruptedException { Flux<String> netFlux = Flux.fromStream(ReactiveJavaTutorial::getVideo) .delayElements(Duration.ofSeconds(2)); // each part will play for 2 seconds // First Subscriber netFlux.subscribe(part -> System.out.println(“Subscriber 1: ” + part)); // wait 5 seconds before next Subscriber joins Thread.sleep(5000); // Seconds Subscriber netFlux.subscribe(part -> System.out.println(“Subscriber 2: ” + part)); Thread.sleep(60000); }…

class ReactiveJavaTutorial { public static void main(String[] args) { Mono.just(“data1”) .concatWith(Flux.error(new RuntimeException(“Exception occurred.”))) .doOnError(ex -> System.out.println(“LOG: Exception caught: ” + ex)) .retry(3) //retry 3 times in case of an error .log() .subscribe(); } } Output: reactor.Flux.Retry.1 : onSubscribe(FluxRetry.RetrySubscriber) reactor.Flux.Retry.1 : request(unbounded) reactor.Flux.Retry.1 : onNext(data1) LOG: Exception caught: java.lang.RuntimeException: Exception occurred. reactor.Flux.Retry.1 : onNext(data1) LOG: Exception…

class ReactiveJavaTutorial { public static void main(String[] args) { Flux.just(2, 7, 10) .concatWith(Flux.error(new RuntimeException(“Exception occurred”))) .concatWith(Mono.just(12)) .log() .subscribe(); } } Output: reactor.Flux.ConcatArray.1 : onSubscribe(FluxConcatArray.ConcatArraySubscriber) reactor.Flux.ConcatArray.1 : request(unbounded) reactor.Flux.ConcatArray.1 : onNext(2) reactor.Flux.ConcatArray.1 : onNext(7) reactor.Flux.ConcatArray.1 : onNext(3) reactor.Flux.ConcatArray.1 : onError(java.lang.RuntimeException: Exception occurred) reactor.Flux.ConcatArray.1 : java.lang.RuntimeException: Exception occurred at com.example.demo.DemoApplication.main(DemoApplication.java:14) ~[classes/:na]   You can see that element…

class ReactiveJavaTutorial { public static void main(String[] args) { Flux.just(“data1”, “data2”, “data3”) .doOnSubscribe(subscription -> System.out.println(“Subscription: ” + subscription)) .subscribe(); } } Output: Subscription: [email protected] doOnNext() method class ReactiveJavaTutorial { public static void main(String[] args) { Flux.just(“data1”, “data2”, “data3”) .doOnNext(data -> System.out.println(“Data: ” + data)) .subscribe(); } } Output: Data: data1 Data: data2 Data: data3 doOnComplete()…

class ReactiveJavaTutorial { public static void main(String[] args) { Flux<String> firstFlux = Flux.fromArray(new String[]{“a”, “b”, “c”}); Flux<String> secondFlux = Flux.fromArray(new String[]{“d”, “e”, “f”}); // combine two Flux Publishers Flux<String> combinedFlux = Flux.concat(firstFlux, secondFlux); // subscribe and print the elements of a combined Flux combinedFlux.subscribe(element -> System.out.print(element + ” “)); } } Output: a b c…

class ReactiveJavaTutorial { public static void main(String[] args) { Flux.fromArray(new String[]{“Tom”, “Melissa”, “Steve”, “Megan”}) .map(String::toUpperCase) .subscribe(System.out::println); } } Output: TOM MELISSA STEVE MEGAN class ReactiveJavaTutorial { public static void main(String[] args) { Flux.fromArray(new String[]{“Tom”, “Melissa”, “Steven”, “Megan”}) .filter(name -> name.length() > 5) .map(String::toUpperCase) .subscribe(System.out::println); } } Output: MELISSA STEVEN   Note: Reactive Streams are immutable.…

Flux<String> cities = Flux.fromIterable( new ArrayList<>(Arrays.asList(“New York”, “London”, “Paris”, “Toronto”, “Rome”))); cities.subscribe(); class ReactiveJavaTutorial { public static void main(String[] args) { Flux<String> cities = Flux.fromIterable( new ArrayList<>(Arrays.asList(“New York”, “London”, “Paris”, “Toronto”, “Rome”))); cities.log().subscribe(); } } Output: INFO 6832 — [ main] reactor.Flux.Iterable.1 : | onSubscribe([Synchronous Fuseable] FluxIterable.IterableSubscription) INFO 6832 — [ main] reactor.Flux.Iterable.1 : |…

class ReactiveJavaTutorial { public static void main(String[] args) { Mono<String> mono = Mono.just(“data”); Flux<String> fluxFromMono = Flux.from(mono); fluxFromMono.subscribe(System.out::println); } } Output: data class ReactiveJavaTutorial { public static void main(String[] args) { // one value Flux<String> flux1 = Flux.just(“data1”); Mono<String> monoFromFlux1 = flux1.next(); // get data from mono monoFromFlux1.subscribe(data -> System.out.println(“monoFromFlux1 data: ” + data)); //…

class ReactiveJavaTutorial { public static void main(String[] args) throws InterruptedException { Flux<String> flux = Flux.fromArray(new String[]{“data1”, “data2”, “data3”}); flux.subscribe(System.out::println); } } Output: data1 data2 data3   Another way would be using the Reactive Streams operators like onNext, flatMap, etc. class ReactiveJavaTutorial { public static void main(String[] args) throws InterruptedException { Flux<String> flux = Flux.fromArray(new String[]{“data1”,…

class ReactiveJavaTutorial { public static void main(String[] args) throws InterruptedException { // Create a Flux Flux<String> flux = Flux.just(“data1”, “data2”, “data3”); // Subscribe to a Flux and print the elements flux.subscribe(System.out::println); } } Output: data1 data2 data3   When we call the subscribe(), we are telling the Publisher to start sending data. class ReactiveJavaTutorial {…

import reactor.core.publisher.Flux; class ReactiveJavaTutorial { public static void main(String[] args) throws InterruptedException { // create an empty Flux Flux flux1 = Flux.just(); // create a Flux that will hold only one value Flux<String> flux2 = Flux.just(“data”); // create a Flux that will hold multiple values Flux<String> flux3 = Flux.just(“data1”, “data2”, “data3”); } } import reactor.core.publisher.Flux;…