Reactive Streams & Reactive Programming in Java

// 1. Publisher — source of data
public interface Publisher<T> {
    void subscribe(Subscriber<? super T> subscriber);
}

// 2. Subscriber — consumer of data
public interface Subscriber<T> {
    void onSubscribe(Subscription subscription);
    void onNext(T item);
    void onError(Throwable throwable);
    void onComplete();
}

// 3. Subscription — link between Publisher and Subscriber
public interface Subscription {
    void request(long n);   // request n items (backpressure!)
    void cancel();          // cancel the subscription
}

// 4. Processor — both Publisher and Subscriber (transforms data in-flight)
public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}

// These are NESTED interfaces inside java.util.concurrent.Flow
Flow.Publisher<T>
Flow.Subscriber<T>
Flow.Subscription
Flow.Processor<T, R>

import java.util.concurrent.Flow.*;
import java.util.concurrent.SubmissionPublisher;

public class FlowExample {
    public static void main(String[] args) throws InterruptedException {
        // SubmissionPublisher is a concrete Publisher
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        // Create a Subscriber
        Subscriber<String> subscriber = new Subscriber<>() {
            private Subscription subscription;

            @Override
            public void onSubscribe(Subscription subscription) {
                this.subscription = subscription;
                subscription.request(1); // request first item
            }

            @Override
            public void onNext(String item) {
                System.out.println("Received: " + item);
                subscription.request(1); // request next item
            }

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

            @Override
            public void onComplete() {
                System.out.println("Done!");
            }
        };

        publisher.subscribe(subscriber);
        publisher.submit("Hello");
        publisher.submit("Reactive");
        publisher.submit("World");
        publisher.close(); // sends onComplete

        Thread.sleep(1000); // wait for async processing
    }
}

Flux.range(1, 1_000_000)
    .onBackpressureBuffer(256)        // buffer up to 256 items
    .publishOn(Schedulers.parallel())
    .subscribe(
        item -> {
            // slow consumer
            Thread.sleep(10);
            System.out.println(item);
        }
    );

// --- Mono creation ---
Mono<String> empty = Mono.empty();
Mono<String> just = Mono.just("Hello");
Mono<String> fromCallable = Mono.fromCallable(() -> fetchFromDb());
Mono<String> fromFuture = Mono.fromFuture(completableFuture);
Mono<String> deferred = Mono.defer(() -> Mono.just(expensiveCall()));

// --- Flux creation ---
Flux<Integer> fromValues = Flux.just(1, 2, 3, 4, 5);
Flux<String> fromIterable = Flux.fromIterable(List.of("a", "b", "c"));
Flux<Integer> range = Flux.range(1, 100);
Flux<Long> interval = Flux.interval(Duration.ofSeconds(1)); // 0, 1, 2, ...

// generate — synchronous, stateful, one-by-one
Flux<String> generated = Flux.generate(
    () -> 0,                           // initial state
    (state, sink) -> {
        sink.next("Item " + state);
        if (state == 9) sink.complete();
        return state + 1;
    }
);

// create — async, multi-threaded, bridge from callback APIs
Flux<String> created = Flux.create(sink -> {
    // Bridge a callback-based API
    myEventSource.register(event -> sink.next(event));
    myEventSource.onClose(() -> sink.complete());
    myEventSource.onError(e -> sink.error(e));
});

Flux<String> names = Flux.just("john", "jane", "bob");

// map — synchronous 1:1 transform
Flux<String> upper = names.map(String::toUpperCase);
// Output: JOHN, JANE, BOB

// flatMap — async 1:N transform (interleaved, unordered)
Flux<User> users = names.flatMap(name -> userService.findByName(name));

// concatMap — async 1:N transform (preserves order, sequential)
Flux<User> orderedUsers = names.concatMap(name -> userService.findByName(name));

// switchMap — cancels previous inner publisher on new item
Flux<SearchResult> results = searchTerms
    .switchMap(term -> searchService.search(term));

Flux.range(1, 20)
    .filter(i -> i % 2 == 0)         // even numbers
    .take(5)                          // first 5
    .skip(1)                          // skip first
    .distinct()                       // deduplicate
    .subscribe(System.out::println);  // 4, 6, 8, 10

Flux<String> flux1 = Flux.just("A", "B", "C");
Flux<Integer> flux2 = Flux.just(1, 2, 3);

// zip — combines element-by-element into tuples
Flux<String> zipped = Flux.zip(flux1, flux2, (s, i) -> s + i);
// Output: A1, B2, C3

// merge — interleaves emissions as they arrive (real-time)
Flux<String> merged = Flux.merge(flux1, flux2.map(String::valueOf));
// Output order depends on timing

// concat — appends one after another (sequential)
Flux<String> concatenated = Flux.concat(flux1, flux2.map(String::valueOf));
// Output: A, B, C, 1, 2, 3

// combineLatest — combines latest from each source
Flux<String> combined = Flux.combineLatest(flux1, flux2, (s, i) -> s + i);

Flux<Integer> withErrors = Flux.just(1, 2, 0, 4)
    .map(i -> 10 / i); // ArithmeticException at 0

// onErrorReturn — provide a fallback value
withErrors
    .onErrorReturn(-1)
    .subscribe(System.out::println); // 10, 5, -1

// onErrorResume — switch to a fallback publisher
withErrors
    .onErrorResume(e -> Flux.just(99, 100))
    .subscribe(System.out::println); // 10, 5, 99, 100

// onErrorMap — transform the error
withErrors
    .onErrorMap(ArithmeticException.class, 
                e -> new BusinessException("Division failed", e))
    .subscribe();

// retry — resubscribe on error
withErrors
    .retry(3) // retry up to 3 times
    .subscribe();

// retryWhen — advanced retry with backoff
withErrors
    .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
        .maxBackoff(Duration.ofSeconds(10))
        .jitter(0.5))
    .subscribe();

// doOnError — side effect (logging) without handling
withErrors
    .doOnError(e -> log.error("Stream failed", e))
    .onErrorReturn(-1)
    .subscribe();

Flux.range(1, 100)
    .subscribeOn(Schedulers.boundedElastic())   // upstream runs here
    .map(i -> blockingDbCall(i))                // runs on boundedElastic
    .publishOn(Schedulers.parallel())            // downstream switches here
    .map(i -> cpuIntensiveTransform(i))          // runs on parallel
    .subscribe();

// COLD — each subscriber triggers a new HTTP call
Mono<User> cold = webClient.get()
    .uri("/users/1")
    .retrieve()
    .bodyToMono(User.class);

cold.subscribe(u -> log.info("Sub1: {}", u)); // HTTP call #1
cold.subscribe(u -> log.info("Sub2: {}", u)); // HTTP call #2

// HOT — convert cold to hot using share()
Flux<Long> hot = Flux.interval(Duration.ofSeconds(1))
    .share(); // multicasts to all current subscribers

hot.subscribe(i -> System.out.println("Sub1: " + i));
Thread.sleep(3000);
hot.subscribe(i -> System.out.println("Sub2: " + i)); // misses first 3 items

// RxJava 3
Flowable.just(1, 2, 3)
    .map(i -> i * 2)
    .filter(i -> i > 2)
    .subscribe(System.out::println);

// Project Reactor (equivalent)
Flux.just(1, 2, 3)
    .map(i -> i * 2)
    .filter(i -> i > 2)
    .subscribe(System.out::println);

// Stream — sync, pull-based, single use
List<String> result = users.stream()
    .filter(u -> u.isActive())
    .map(User::getName)
    .collect(Collectors.toList());

// CompletableFuture — async, single value, eager
CompletableFuture<User> future = CompletableFuture
    .supplyAsync(() -> userService.findById(1))
    .thenApply(user -> enrich(user));

// Mono — async, single value, lazy, composable
Mono<User> mono = userService.findById(1)
    .map(user -> enrich(user))
    .onErrorResume(e -> Mono.empty());

// Testing a Flux
@Test
void testFluxEmitsCorrectValues() {
    Flux<String> source = Flux.just("Alpha", "Beta", "Gamma");

    StepVerifier.create(source)
        .expectNext("Alpha")
        .expectNext("Beta")
        .expectNext("Gamma")
        .expectComplete()       // assert onComplete signal
        .verify();              // triggers the subscription
}

// Testing a Mono with error
@Test
void testMonoError() {
    Mono<String> errorMono = Mono.error(new RuntimeException("Oops"));

    StepVerifier.create(errorMono)
        .expectErrorMatches(e -> e instanceof RuntimeException 
                                 && e.getMessage().equals("Oops"))
        .verify();
}

// Testing with virtual time (for time-based operators)
@Test
void testIntervalWithVirtualTime() {
    StepVerifier.withVirtualTime(() -> 
            Flux.interval(Duration.ofHours(1)).take(3))
        .expectSubscription()
        .thenAwait(Duration.ofHours(3))   // advance virtual clock
        .expectNext(0L, 1L, 2L)
        .expectComplete()
        .verify();
}

// Testing backpressure
@Test
void testBackpressure() {
    Flux<Integer> source = Flux.range(1, 100);

    StepVerifier.create(source, 5)  // request only 5
        .expectNext(1, 2, 3, 4, 5)
        .thenCancel()               // cancel after receiving 5
        .verify();
}

// Testing with context
@Test
void testReactorContext() {
    Mono<String> greeting = Mono.deferContextual(ctx ->
        Mono.just("Hello, " + ctx.get("user"))
    );

    StepVerifier.create(greeting.contextWrite(ctx -> ctx.put("user", "Vamsi")))
        .expectNext("Hello, Vamsi")
        .expectComplete()
        .verify();
}

// Call multiple services in parallel and merge results
Mono<UserProfile> profile = userService.getProfile(userId);
Mono<List<Order>> orders = orderService.getOrders(userId);
Mono<Preferences> prefs = prefService.getPreferences(userId);

Mono<Dashboard> dashboard = Mono.zip(profile, orders, prefs)
    .map(tuple -> new Dashboard(
        tuple.getT1(),  // profile
        tuple.getT2(),  // orders
        tuple.getT3()   // preferences
    ));

Mono<Response> resilientCall = webClient.get()
    .uri("/external-api/data")
    .retrieve()
    .bodyToMono(Response.class)
    .timeout(Duration.ofSeconds(3))
    .retryWhen(Retry.backoff(3, Duration.ofMillis(500))
        .filter(e -> e instanceof WebClientResponseException.ServiceUnavailable)
        .onRetryExhaustedThrow((spec, signal) -> 
            new ServiceUnavailableException("External API down after retries")))
    .onErrorResume(e -> Mono.just(Response.fallback()));

// Process a list of items in parallel with controlled concurrency
Flux<Result> results = Flux.fromIterable(itemIds)
    .flatMap(
        id -> processItem(id),    // async processing per item
        16                        // concurrency limit (max 16 in parallel)
    )
    .collectList()                // fan-in: gather all results
    .flatMapMany(Flux::fromIterable);