Programmation réactive avec Reactor

Mono<String> mono = Mono.just("Hello, Reactor!");
mono.subscribe(System.out::println);

Flux<Integer> flux = Flux.range(1, 5);
flux.subscribe(System.out::println);

public class BackpressureSubscriber<T> implements Subscriber<T> {
    private Subscription subscription;
    private final int batchSize = 3;
    private int currentReceived = 0;
    private static final int PROCESSING_TIME_MS = 1000;

    @Override
    public void onSubscribe(Subscription s) {
        this.subscription = s;
        System.out.println("🔵 Initial request for " + batchSize + " items");
        subscription.request(batchSize); // Demande initiale
    }

    @Override
    public void onNext(T item) {
        try {
            Thread.sleep(PROCESSING_TIME_MS); // Traitement lent
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.println("📥 Received: " + item);
        currentReceived++;
        
        if (currentReceived == batchSize) {
            currentReceived = 0;
            System.out.println("🔄 Requesting next " + batchSize + " items");
            subscription.request(batchSize);
        }
    }

    @Override
    public void onError(Throwable t) {
        System.err.println("❌ Error: " + t.getMessage());
    }

    @Override
    public void onComplete() {
        System.out.println("✅ Completed");
    }
}

Flux<Integer> fluxWithResume = Flux.range(1, 5)
                                   .map(i -> {
                                       if (i == 3) {
                                           throw new RuntimeException("Erreur à l'élément 3");
                                       }
                                       return i;
                                   })
                                   // Remplace par un autre flux en cas d'erreur
                                   .onErrorResume(e -> Flux.just(10, 20, 30)); 
fluxWithResume.subscribe(System.out::println);

import java.time.Duration;
Flux<Long> bufferedFlux = Flux.interval(Duration.ofMillis(10)) // Émission très rapide
                                 // Stocke les éléments dans un buffer de taille 10 en cas de backpressure
                                 .onBackpressureBuffer(10, i -> System.out.println("Buffered: " + i))
                                 .take(50);
bufferedFlux.subscribe(new BackpressureSubscriber());

System.out.println("Number of send elements: "+bufferedFlux.blockLast()); // Bloque l'exécution pour voir le résultat

Flux<Long> latestFlux = Flux.interval(Duration.ofMillis(10)) // Émission très rapide
                        .onBackpressureLatest() // Garde uniquement le dernier élément en cas de backpressure                        
                        .take(50);

latestFlux.subscribe(new BackpressureSubscriber());

latestFlux.blockLast();

%maven io.projectreactor:reactor-test:3.7.0
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;

Flux<String> testedFlux = Flux.just("A", "B", "C");

StepVerifier.create(testedFlux)
            .expectNext("A")
            .expectNext("B")
            .expectNext("C")
            .expectNext("D") // Erreur attendue
            .verifyComplete();

Flux<Integer> flux = Flux.range(1, 2)
                         .log(); // Ajoute des logs pour chaque événement du flux

flux.subscribe(System.out::println);

Flux<Integer> flux = Flux.range(1, 5)
                         .map(i -> {
                             if (i == 3) {
                                 throw new RuntimeException("Erreur à l'élément 3");
                             }
                             return i;
                         })
                         .checkpoint("Point de contrôle avant l'erreur");

flux.subscribe(System.out::println, Throwable::printStackTrace);