Reactive Programming with Project Reactor: Mono and Flux

//CreatingStreams.java
package com.javabrahman.reactive.basics;

import java.util.List;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class CreatingStreams {

  static String load(String isbn) {
    System.out.println("   loading " + isbn + " from the database");
    return "Effective Java";
  }

  public static void main(String[] args) {
    Mono<String> fixed = Mono.just("Domain-Driven Design");               // a value that exists already
    Mono<String> lazy = Mono.fromSupplier(() -> load("111"));   // computed on subscription
    Flux<Integer> numbers = Flux.range(1, 5);
    Flux<String> titles = Flux.fromIterable(List.of("Refactoring", "Clean Code"));
    Mono<String> nothing = Mono.empty();
    Flux<String> failing = Flux.error(new IllegalStateException("disk full"));

    System.out.println("1. pipelines created, nothing has run yet");

    System.out.println("2. subscribing to lazy:");
    lazy.subscribe(title -> System.out.println("   got " + title));
    System.out.println("   and again - the supplier runs once per subscriber:");
    lazy.subscribe(title -> System.out.println("   got " + title));

    System.out.println("3. a Flux, with handlers for all three kinds of signal:");
    numbers.map(n -> n * n).subscribe(
        n -> System.out.println("   onNext " + n),
        error -> System.out.println("   onError " + error),
        () -> System.out.println("   onComplete"));

    System.out.println("4. empty, an error, and a fixed value:");
    nothing.subscribe(t -> System.out.println("   never printed"), e -> { }, () -> System.out.println("   empty: onComplete only"));
    failing.subscribe(t -> { }, e -> System.out.println("   failing: onError " + e.getMessage()));
    titles.concatWith(fixed).subscribe(t -> System.out.println("   " + t));
  }
}

//Operators.java
package com.javabrahman.reactive.basics;

import java.time.Duration;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class Operators {

  // stands for a non-blocking call that answers after some time
  static Flux<String> reviewsOf(String book, int delayMillis) {
    return Flux.just(book + " review 1", book + " review 2").delayElements(Duration.ofMillis(delayMillis));
  }

  public static void main(String[] args) {
    System.out.println("map, filter, take, collectList:");
    Flux.range(1, 10)
        .filter(n -> n % 2 == 0)
        .map(n -> "#" + n)
        .take(3)
        .collectList()                          // Flux<String> -> Mono<List<String>>
        .subscribe(list -> System.out.println("   " + list));

    System.out.println("flatMap - inner publishers run at the same time, results interleave:");
    Flux.just("A", "B")
        .flatMap(book -> reviewsOf(book, book.equals("A") ? 300 : 100))
        .doOnNext(r -> System.out.println("   " + r))
        .blockLast();                           // wait - only acceptable in a main method or a test

    System.out.println("concatMap - one inner publisher after the other, in order:");
    Flux.just("A", "B")
        .concatMap(book -> reviewsOf(book, book.equals("A") ? 300 : 100))
        .doOnNext(r -> System.out.println("   " + r))
        .blockLast();

    System.out.println("zip - combine the results of independent calls:");
    Mono<String> title = Mono.just("Effective Java").delayElement(Duration.ofMillis(200));
    Mono<Integer> price = Mono.just(450).delayElement(Duration.ofMillis(200));
    long start = System.nanoTime();
    String combined = Mono.zip(title, price, (t, p) -> t + " costs Rs. " + p).block();
    System.out.println("   " + combined + ", after about " + Math.round((System.nanoTime() - start) / 100_000_000.0) * 100 + " ms");
  }
}

//Errors.java
package com.javabrahman.reactive.basics;

import java.util.concurrent.atomic.AtomicInteger;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class Errors {

  public static void main(String[] args) {
    System.out.println("an exception in map ends the stream:");
    Flux.just(10, 5, 0, 2)
        .map(n -> 100 / n)
        .subscribe(n -> System.out.println("   " + n), e -> System.out.println("   onError " + e));

    System.out.println("onErrorReturn - a fallback value, then completion:");
    Flux.just(10, 5, 0, 2)
        .map(n -> 100 / n)
        .onErrorReturn(-1)
        .subscribe(n -> System.out.println("   " + n), e -> { }, () -> System.out.println("   onComplete"));

    System.out.println("onErrorResume - a fallback publisher, chosen by the error:");
    Mono.<String>error(new IllegalStateException("price service down"))
        .onErrorResume(IllegalStateException.class, e -> Mono.just("price from the cache"))
        .subscribe(s -> System.out.println("   " + s));

    System.out.println("onErrorMap - translate the exception:");
    Mono.error(new IllegalArgumentException("bad isbn"))
        .onErrorMap(e -> new IllegalStateException("lookup failed: " + e.getMessage()))
        .subscribe(s -> { }, e -> System.out.println("   " + e));

    System.out.println("retry - subscribe again, here up to 2 more times:");
    AtomicInteger attempts = new AtomicInteger();
    Mono.fromCallable(() -> {
          int n = attempts.incrementAndGet();
          System.out.println("   attempt " + n);
          if (n < 3) {
            throw new IllegalStateException("timeout");
          }
          return "rate 88.5";
        })
        .retry(2)
        .doFinally(signal -> System.out.println("   doFinally: " + signal))
        .subscribe(s -> System.out.println("   " + s));
  }
}

//Threads.java
package com.javabrahman.reactive.basics;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

public class Threads {

  static String where(String step) {
    return "   " + step + " on " + Thread.currentThread().getName().replaceAll("-\\d+$", "-N");
  }

  // a blocking call, such as JDBC: must not run on a reactive thread
  static String blockingLookup() throws InterruptedException {
    System.out.println(where("blocking lookup"));
    Thread.sleep(100);
    return "Effective Java";
  }

  public static void main(String[] args) {
    System.out.println("no scheduler - everything on the subscribing thread:");
    Flux.range(1, 1).map(n -> where("map")).subscribe(System.out::println);

    System.out.println("publishOn - switches for the operators after it:");
    Flux.range(1, 1)
        .map(n -> where("map before publishOn"))
        .doOnNext(System.out::println)
        .publishOn(Schedulers.parallel())
        .map(s -> where("map after publishOn"))
        .doOnNext(System.out::println)
        .blockLast();

    System.out.println("subscribeOn - moves the source, wherever it stands:");
    Mono.fromCallable(Threads::blockingLookup)
        .map(title -> where("map"))
        .subscribeOn(Schedulers.boundedElastic())
        .doOnNext(System.out::println)
        .block();
  }
}

//Backpressure.java
package com.javabrahman.reactive.basics;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;

public class Backpressure {

  public static void main(String[] args) {
    System.out.println("a subscriber that takes 2 elements at a time:");
    Flux.range(1, 5)
        .doOnRequest(n -> System.out.println("   request(" + n + ")"))
        .subscribe(new BaseSubscriber<Integer>() {
          private int received;

          @Override
          protected void hookOnSubscribe(Subscription subscription) {
            request(2);
          }

          @Override
          protected void hookOnNext(Integer value) {
            System.out.println("   onNext " + value);
            if (++received % 2 == 0) {
              request(2);   // ready for the next batch
            }
          }

          @Override
          protected void hookOnComplete() {
            System.out.println("   onComplete");
          }
        });

    System.out.println("limitRate(3) - an unbounded subscriber, but requests upstream in batches:");
    Flux.range(1, 7)
        .doOnRequest(n -> System.out.println("   request(" + n + ")"))
        .limitRate(3)
        .subscribe(n -> System.out.println("   onNext " + n));
  }
}