Kotlin Spring Reactivity: Practical Course

Senior-level focus: the shift from Thread-per-Request (Spring MVC) to the Event Loop (WebFlux). Master Mono, Flux, map vs flatMap, functional routing, R2DBC, and Reactive MongoDB — with Kotlin and hands-on challenges.

Course contents: Module 1 — Reactive Mindset. Module 2 — Reactor (map, flatMap, zip, Schedulers). Module 3 — WebFlux functional routing. Module 4 — Reactive data (R2DBC + Mongo). Code challenges with editable snippets. Senior challenges: Rate Limiter, Multi-Source, SSE, Anti-Pattern Hunt.

Module 1: The Reactive Mindset

1.1 Why Blocking Kills WebFlux

In a standard Servlet container (Tomcat), 200 threads all waiting on a slow DB means the server is effectively dead. In WebFlux you use a small number of threads (often equal to CPU cores). If you block one, you freeze a huge chunk of your app. Non-blocking I/O is mandatory.

1.2 The “Nothing Happens Until You Subscribe” Rule

The #1 mistake: writing a pipeline and wondering why nothing runs. Reactive streams are lazy — they only execute when there is a subscribe.

Bad (imperative mindset):

kotlin

// BAD (Imperative): This just defines a recipe — it doesn't run!
fun updateData() {
    Mono.just("New Data")
        .map { it.uppercase() }
        .doOnNext { println(it) }  // Nothing is printed — no subscribe!
}

Good (reactive mindset):

kotlin

// GOOD (Reactive): Subscribe executes the pipeline
fun updateData() {
    Mono.just("New Data")
        .map { it.uppercase() }
        .subscribe { println(it) }  // Now the pipeline runs
}

kotlin

import reactor.core.publisher.Mono
import reactor.kotlin.core.publisher.toMono

fun main() {
    val simpleMono: Mono<String> = Mono.just("Hello World")
    val name = "Christian"
    val kotlinMono: Mono<String> = name.toMono()
    val safeMono = Mono.justOrEmpty(null as String?)
    simpleMono.subscribe { println(it) }
}

Module 2: Reactor Basics (Operators)

2.1 Map vs FlatMap

map: synchronous transformation (value → value). flatMap: asynchronous transformation when the step returns Mono or Flux (e.g. a DB call). Use flatMap for I/O.

kotlin

// map = synchronous transformation (value in, value out)
val parsedIds: Flux<Int> = Flux.just("1", "2", "3").map { it.toInt() }

// flatMap = async transformation (e.g. DB call returns Mono/Flux)
val users: Flux<User> = Flux.just("1", "2", "3")
    .flatMap { id -> userRepository.findById(id) }

2.2 Zip vs FlatMap

zip: use when you need all sources to finish to build one object (e.g. User + Account + Profile). flatMap: when one call depends on the previous (e.g. get User then get Orders by userId).

kotlin

// zip: run all sources in parallel, combine when all complete
fun getDashboard(userId: String): Mono<UserDashboard> {
    return Mono.zip(
        userService.findById(userId),
        orderService.getRecentOrder(userId),
        creditService.getBalance(userId)
    ).map { tuple ->
        UserDashboard(tuple.t1, tuple.t2, tuple.t3)
    }
}

2.3 Schedulers

The event-loop thread must never block. For legacy blocking calls, move work to Schedulers.boundedElastic(). subscribeOn affects the source; publishOn affects operators after it.

kotlin

// Event-loop thread must never block. Offload blocking work:
fun callLegacySystem(): Mono<String> {
    return Mono.fromCallable { legacyBlockingCall() }
        .subscribeOn(Schedulers.boundedElastic())
}

// subscribeOn = where the SOURCE runs
// publishOn = where operators AFTER it run

kotlin

import reactor.core.publisher.Flux

val numbersFlux: Flux<Int> = Flux.just(1, 2, 3, 4, 5)
val productFlux = Flux.fromIterable(listOf("TV", "Phone", "Laptop"))
productFlux.map { it.uppercase() }.filter { it.length > 2 }.subscribe { println(it) }

Module 3: Spring WebFlux Fundamentals

Functional routing is often preferred over @RestController for testability and explicitness. You define a Router (routes) and a Handler (logic returning Mono<ServerResponse>).

Router:

kotlin

@Configuration
class OrderRouter {
    @Bean
    fun route(handler: OrderHandler): RouterFunction<ServerResponse> =
        RouterFunctions
            .route(GET("/orders/{id}"), handler::getOrder)
            .andRoute(GET("/orders").and(accept(APPLICATION_JSON)), handler::listOrders)
}

Handler:

kotlin

@Component
class OrderHandler(private val orderService: OrderService) {
    fun getOrder(request: ServerRequest): Mono<ServerResponse> {
        val id = request.pathVariable("id")
        return ServerResponse.ok()
            .body(orderService.findById(id), Order::class.java)
    }
}

Module 4: The Reactive Data Layer

Reactive drivers don’t hold a thread while waiting for the DB; they register a callback on the I/O socket and free the thread. R2DBC for Postgres (relational); MongoDB Reactive for document stores and tailable cursors (e.g. SSE).

R2DBC (PostgreSQL)

kotlin

@Repository
interface ProductRepository : ReactiveSortingRepository<Product, Long> {
    fun findByCategory(category: String): Flux<Product>
}

MongoDB Reactive

kotlin

interface OrderRepository : ReactiveMongoRepository<Order, String> {
    @Tailable
    fun findByStatus(status: String): Flux<Order>
}

Transactions

In WebFlux, @Transactional still works; context is handled via Reactor Context instead of ThreadLocal.

kotlin

@Service
class InventoryService(private val repository: ProductRepository) {
    @Transactional
    fun updateStock(id: Long, quantity: Int): Mono<Void> {
        return repository.findById(id)
            .flatMap { p ->
                p.stock = p.stock - quantity
                repository.save(p)
            }
            .then()
    }
}

N+1 in Reactive: Wrong vs Right

Using .block() inside a map forces sequential, blocking behavior and kills the event loop. Use flatMap to keep non-blocking and concurrent.

Wrong:

kotlin

// BAD: .block() kills the event loop — sequential, blocking
fun getOrdersBad(): Flux<OrderDTO> {
    return orderRepository.findAll()
        .map { order ->
            val user = userRepository.findById(order.userId).block()!!
            OrderDTO(order, user)
        }
}

Right:

kotlin

// GOOD: flatMap keeps everything non-blocking and concurrent
fun getOrdersGood(): Flux<OrderDTO> {
    return orderRepository.findAll()
        .flatMap { order ->
            userRepository.findById(order.userId)
                .map { user -> OrderDTO(order, user) }
        }
}

Hybrid (Postgres + Mongo): Fetch users from Postgres and activity logs from Mongo, then combine with flatMap + collectList.

kotlin

// Postgres: users. Mongo: activity logs. Unified Flux.
fun getFullAuditLog(): Flux<UserActivity> {
    return postgresRepo.findAllUsers()
        .flatMap { user ->
            mongoRepo.findByUserId(user.id)
                .collectList()
                .map { logs -> UserActivity(user, logs) }
        }
}

Code Challenges: Modules 1 & 2

Challenge 1: The Ghost Subscriber. The method should fetch a user, log their name, and save. Fix it so that the “Saving user…” message actually appears (subscribe and/or doOnNext).

kotlin

// Challenge 1: The Ghost Subscriber
// Fix so "Saving user..." actually appears. (Hint: subscribe!)

fun processUser(id: String) {
    val userMono = Mono.just("Engineer_User")
        .doOnNext { name -> println("Processing: $name") }
    userMono.map { name -> "Saved: $name" }
    // TODO: Make the pipeline execute and log "Saving user..."
}

Challenge 2: Concurrency Bottleneck. List of 5 product IDs. Call productService.getDetails(id) for each (returns Mono). Use flatMap, collectList, then print the result.

kotlin

// Challenge 2: Concurrency Bottleneck
// 1. Flux of IDs: "p1", "p2", "p3"
// 2. Call productService.getDetails(id) for each (returns Mono<Product>)
// 3. Collect into a list and print

val ids = Flux.just("p1", "p2", "p3")
// Your code: flatMap -> getDetails -> collectList -> subscribe(println)

Challenges: Sequential vs Parallel, Zip Refactor

Challenge 3: 10 OrderIDs. (1) Fetch orders one-by-one with concatMap. (2) Fetch concurrently with flatMap. If each call takes 100ms, what’s the total time for each?

kotlin

// Challenge 3: Sequential vs Parallel
// 10 OrderIDs — fetch Order for each.
// (1) Sequential: concatMap — one by one
// (2) Concurrent: flatMap — in parallel
// If each DB call takes 100ms, total time?

val orderIds = Flux.range(1, 10).map { "order-$it" }
// Sequential: orderIds.concatMap { id -> orderRepo.findById(id) }
// Parallel:   orderIds.flatMap { id -> orderRepo.findById(id) }

Challenge 4: Zip Refactor. Run productRepo.findById(id) and reviewRepo.findAllByProductId(id).collectList() in parallel with Mono.zip, then map to ProductDetails.

kotlin

// Challenge 4: Zip Refactor
// Run product + reviews in PARALLEL (zip), not sequential (flatMap).

fun getDetails(id: String): Mono<ProductDetails> {
    return productRepo.findById(id)
        .flatMap { product ->
            reviewRepo.findAllByProductId(id).collectList()
                .map { reviews -> ProductDetails(product, reviews) }
        }
}
// Refactor to: Mono.zip(productRepo.findById(id), reviewRepo.findAllByProductId(id).collectList())
//             .map { (product, reviews) -> ProductDetails(product, reviews) }

Senior Challenges

Challenge 5: Smart Rate Limiter. Flux of 100 IDs; call a legacy API (max 5 concurrent). Use flatMap(..., 5). Bonus: retry 3 times with exponential backoff (1s).

kotlin

// Challenge 5: Smart Rate Limiter
// 100 request IDs; call dependencyService.call(id); max 5 concurrent.
// Bonus: retry 3 times with exponential backoff (1s) on failure.

Flux.range(1, 100)
    .flatMap({ id -> dependencyService.call(id) }, 5)  // concurrency = 5
    .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))

Challenge 6: Multi-Source Aggregator. User from Postgres, Preferences from Mongo. Fetch both in parallel with Mono.zip. User not found → error; Preferences not found → default.

kotlin

// Challenge 6: Multi-Source Aggregator (Postgres + Mongo)
// User in Postgres, Preferences in Mongo. Fetch BOTH in parallel.
// If user not in Postgres -> UserNotFoundException
// If prefs not in Mongo -> default preferences

fun getProfile(userId: String): Mono<FullProfile> {
    val user = postgresRepo.findUserById(userId)
        .switchIfEmpty(Mono.error(UserNotFoundException(userId)))
    val prefs = mongoRepo.findPreferences(userId).defaultIfEmpty(DefaultPreferences())
    return Mono.zip(user, prefs).map { (u, p) -> FullProfile(u, p) }
}

Challenge: Infinite Activity Stream (SSE). Create GET /logs/stream with a functional router. Use a @Tailable cursor on a MongoDB capped collection; merge with a 2-second heartbeat (e.g. Flux.interval) so the connection stays alive. Respond with MediaType.TEXT_EVENT_STREAM. Test with StepVerifier for the first 3 signals.

kotlin

// GET /logs/stream — SSE from MongoDB @Tailable cursor + heartbeat every 2s
@Bean
fun logStreamRoute(handler: LogStreamHandler): RouterFunction<ServerResponse> =
    RouterFunctions.route(GET("/logs/stream"), handler::stream)

// Handler: logRepo.findByStatus("ACTIVE") is @Tailable Flux
// Merge with Flux.interval(Duration.ofSeconds(2)).map { "" } for heartbeat
// MediaType.TEXT_EVENT_STREAM

Challenge 7: Anti-Pattern Hunt. Identify and fix: (A) .block() inside a stream, (B) Thread.sleep, (C) wrong scheduler. Refactor to a fully non-blocking pipeline.

kotlin

// Challenge 7: Anti-Pattern Hunt — fix the 3 reactive sins
// A: .block() — kills event loop
// B: Thread.sleep — blocks thread
// C: subscribeOn(immediate()) — wrong scheduler

fun processOrders(): Flux<OrderResponse> {
    return orderRepository.findAll()
        .flatMap { order ->
            userRepository.findById(order.userId)
                .map { user -> OrderResponse(order, user) }
        }
    // Remove .block(), Thread.sleep, use boundedElastic if you had blocking
}

Capstone: E-Commerce Checkout

Inputs: userId: String, cartIds: List<String>. Validate user, get balance, compute cart total (e.g. Flux.fromIterable(cartIds).flatMap(getItemPrice).reduce(0.0, Double::plus)), then return Mono<String> “Approved” or “Rejected”.

Use Mono.justOrEmpty / switchIfEmpty for validation, flatMap and zipWith to combine balance and total, then map to the result string.

kotlin

// Capstone: E-Commerce Checkout
// userId + cartIds -> balance >= total ? "Approved" : "Rejected"
fun checkout(userId: String, cartIds: List<String>): Mono<String> {
    return Mono.justOrEmpty(userId).filter { it.isNotBlank() }
        .switchIfEmpty(Mono.error(IllegalArgumentException("Invalid User")))
        .flatMap { id -> getUserBalance(id) }
        .zipWith(Flux.fromIterable(cartIds).flatMap { getItemPrice(it) }.reduce(0.0, Double::plus))
        .map { (balance, total) -> if (balance >= total) "Approved" else "Rejected" }
}

Recap: Key Takeaways

Blocking one thread in WebFlux freezes a large part of your capacity; never block the event loop.
Nothing runs until you subscribe. Describe the pipeline, then subscribe.
map = synchronous (value → value). flatMap = async (value → Mono/Flux).
zip = parallel independent calls; flatMap = dependent async steps.
Use Schedulers.boundedElastic() for legacy blocking code; never block on the default scheduler.
Functional Router + Handler: testable, explicit WebFlux endpoints.
R2DBC (Postgres) and Reactive Mongo; avoid N+1 by using flatMap instead of map + .block().

Kotlin Spring Reactivity: Practical Course

Module 1: The Reactive Mindset

1.1 Why Blocking Kills WebFlux

1.2 The “Nothing Happens Until You Subscribe” Rule

The #1 mistake: writing a pipeline and wondering why nothing runs. Reactive streams are lazy — they only execute when there is a subscribe.

Bad (imperative mindset):

kotlin

// BAD (Imperative): This just defines a recipe — it doesn't run!
fun updateData() {
    Mono.just("New Data")
        .map { it.uppercase() }
        .doOnNext { println(it) }  // Nothing is printed — no subscribe!
}

Good (reactive mindset):

kotlin

// GOOD (Reactive): Subscribe executes the pipeline
fun updateData() {
    Mono.just("New Data")
        .map { it.uppercase() }
        .subscribe { println(it) }  // Now the pipeline runs
}

kotlin

import reactor.core.publisher.Mono
import reactor.kotlin.core.publisher.toMono

fun main() {
    val simpleMono: Mono<String> = Mono.just("Hello World")
    val name = "Christian"
    val kotlinMono: Mono<String> = name.toMono()
    val safeMono = Mono.justOrEmpty(null as String?)
    simpleMono.subscribe { println(it) }
}

Module 2: Reactor Basics (Operators)

2.1 Map vs FlatMap

map: synchronous transformation (value → value). flatMap: asynchronous transformation when the step returns Mono or Flux (e.g. a DB call). Use flatMap for I/O.

kotlin

// map = synchronous transformation (value in, value out)
val parsedIds: Flux<Int> = Flux.just("1", "2", "3").map { it.toInt() }

// flatMap = async transformation (e.g. DB call returns Mono/Flux)
val users: Flux<User> = Flux.just("1", "2", "3")
    .flatMap { id -> userRepository.findById(id) }

2.2 Zip vs FlatMap

kotlin

// zip: run all sources in parallel, combine when all complete
fun getDashboard(userId: String): Mono<UserDashboard> {
    return Mono.zip(
        userService.findById(userId),
        orderService.getRecentOrder(userId),
        creditService.getBalance(userId)
    ).map { tuple ->
        UserDashboard(tuple.t1, tuple.t2, tuple.t3)
    }
}

2.3 Schedulers

The event-loop thread must never block. For legacy blocking calls, move work to Schedulers.boundedElastic(). subscribeOn affects the source; publishOn affects operators after it.

kotlin

// Event-loop thread must never block. Offload blocking work:
fun callLegacySystem(): Mono<String> {
    return Mono.fromCallable { legacyBlockingCall() }
        .subscribeOn(Schedulers.boundedElastic())
}

// subscribeOn = where the SOURCE runs
// publishOn = where operators AFTER it run

kotlin

import reactor.core.publisher.Flux

val numbersFlux: Flux<Int> = Flux.just(1, 2, 3, 4, 5)
val productFlux = Flux.fromIterable(listOf("TV", "Phone", "Laptop"))
productFlux.map { it.uppercase() }.filter { it.length > 2 }.subscribe { println(it) }

Module 3: Spring WebFlux Fundamentals

Functional routing is often preferred over @RestController for testability and explicitness. You define a Router (routes) and a Handler (logic returning Mono<ServerResponse>).

Router:

kotlin

@Configuration
class OrderRouter {
    @Bean
    fun route(handler: OrderHandler): RouterFunction<ServerResponse> =
        RouterFunctions
            .route(GET("/orders/{id}"), handler::getOrder)
            .andRoute(GET("/orders").and(accept(APPLICATION_JSON)), handler::listOrders)
}

Handler:

kotlin

@Component
class OrderHandler(private val orderService: OrderService) {
    fun getOrder(request: ServerRequest): Mono<ServerResponse> {
        val id = request.pathVariable("id")
        return ServerResponse.ok()
            .body(orderService.findById(id), Order::class.java)
    }
}

Module 4: The Reactive Data Layer

R2DBC (PostgreSQL)

kotlin

@Repository
interface ProductRepository : ReactiveSortingRepository<Product, Long> {
    fun findByCategory(category: String): Flux<Product>
}

MongoDB Reactive

kotlin

interface OrderRepository : ReactiveMongoRepository<Order, String> {
    @Tailable
    fun findByStatus(status: String): Flux<Order>
}

Transactions

In WebFlux, @Transactional still works; context is handled via Reactor Context instead of ThreadLocal.

kotlin

@Service
class InventoryService(private val repository: ProductRepository) {
    @Transactional
    fun updateStock(id: Long, quantity: Int): Mono<Void> {
        return repository.findById(id)
            .flatMap { p ->
                p.stock = p.stock - quantity
                repository.save(p)
            }
            .then()
    }
}

N+1 in Reactive: Wrong vs Right

Using .block() inside a map forces sequential, blocking behavior and kills the event loop. Use flatMap to keep non-blocking and concurrent.

Wrong:

kotlin

// BAD: .block() kills the event loop — sequential, blocking
fun getOrdersBad(): Flux<OrderDTO> {
    return orderRepository.findAll()
        .map { order ->
            val user = userRepository.findById(order.userId).block()!!
            OrderDTO(order, user)
        }
}

Right:

kotlin

// GOOD: flatMap keeps everything non-blocking and concurrent
fun getOrdersGood(): Flux<OrderDTO> {
    return orderRepository.findAll()
        .flatMap { order ->
            userRepository.findById(order.userId)
                .map { user -> OrderDTO(order, user) }
        }
}

Hybrid (Postgres + Mongo): Fetch users from Postgres and activity logs from Mongo, then combine with flatMap + collectList.

kotlin

// Postgres: users. Mongo: activity logs. Unified Flux.
fun getFullAuditLog(): Flux<UserActivity> {
    return postgresRepo.findAllUsers()
        .flatMap { user ->
            mongoRepo.findByUserId(user.id)
                .collectList()
                .map { logs -> UserActivity(user, logs) }
        }
}

Code Challenges: Modules 1 & 2

Challenge 1: The Ghost Subscriber. The method should fetch a user, log their name, and save. Fix it so that the “Saving user…” message actually appears (subscribe and/or doOnNext).

kotlin

// Challenge 1: The Ghost Subscriber
// Fix so "Saving user..." actually appears. (Hint: subscribe!)

fun processUser(id: String) {
    val userMono = Mono.just("Engineer_User")
        .doOnNext { name -> println("Processing: $name") }
    userMono.map { name -> "Saved: $name" }
    // TODO: Make the pipeline execute and log "Saving user..."
}

Challenge 2: Concurrency Bottleneck. List of 5 product IDs. Call productService.getDetails(id) for each (returns Mono). Use flatMap, collectList, then print the result.

kotlin

// Challenge 2: Concurrency Bottleneck
// 1. Flux of IDs: "p1", "p2", "p3"
// 2. Call productService.getDetails(id) for each (returns Mono<Product>)
// 3. Collect into a list and print

val ids = Flux.just("p1", "p2", "p3")
// Your code: flatMap -> getDetails -> collectList -> subscribe(println)

Challenges: Sequential vs Parallel, Zip Refactor

Challenge 3: 10 OrderIDs. (1) Fetch orders one-by-one with concatMap. (2) Fetch concurrently with flatMap. If each call takes 100ms, what’s the total time for each?

kotlin

// Challenge 3: Sequential vs Parallel
// 10 OrderIDs — fetch Order for each.
// (1) Sequential: concatMap — one by one
// (2) Concurrent: flatMap — in parallel
// If each DB call takes 100ms, total time?

val orderIds = Flux.range(1, 10).map { "order-$it" }
// Sequential: orderIds.concatMap { id -> orderRepo.findById(id) }
// Parallel:   orderIds.flatMap { id -> orderRepo.findById(id) }

Challenge 4: Zip Refactor. Run productRepo.findById(id) and reviewRepo.findAllByProductId(id).collectList() in parallel with Mono.zip, then map to ProductDetails.

kotlin

// Challenge 4: Zip Refactor
// Run product + reviews in PARALLEL (zip), not sequential (flatMap).

fun getDetails(id: String): Mono<ProductDetails> {
    return productRepo.findById(id)
        .flatMap { product ->
            reviewRepo.findAllByProductId(id).collectList()
                .map { reviews -> ProductDetails(product, reviews) }
        }
}
// Refactor to: Mono.zip(productRepo.findById(id), reviewRepo.findAllByProductId(id).collectList())
//             .map { (product, reviews) -> ProductDetails(product, reviews) }

Senior Challenges

Challenge 5: Smart Rate Limiter. Flux of 100 IDs; call a legacy API (max 5 concurrent). Use flatMap(..., 5). Bonus: retry 3 times with exponential backoff (1s).

kotlin

// Challenge 5: Smart Rate Limiter
// 100 request IDs; call dependencyService.call(id); max 5 concurrent.
// Bonus: retry 3 times with exponential backoff (1s) on failure.

Flux.range(1, 100)
    .flatMap({ id -> dependencyService.call(id) }, 5)  // concurrency = 5
    .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))

Challenge 6: Multi-Source Aggregator. User from Postgres, Preferences from Mongo. Fetch both in parallel with Mono.zip. User not found → error; Preferences not found → default.

kotlin

// Challenge 6: Multi-Source Aggregator (Postgres + Mongo)
// User in Postgres, Preferences in Mongo. Fetch BOTH in parallel.
// If user not in Postgres -> UserNotFoundException
// If prefs not in Mongo -> default preferences

fun getProfile(userId: String): Mono<FullProfile> {
    val user = postgresRepo.findUserById(userId)
        .switchIfEmpty(Mono.error(UserNotFoundException(userId)))
    val prefs = mongoRepo.findPreferences(userId).defaultIfEmpty(DefaultPreferences())
    return Mono.zip(user, prefs).map { (u, p) -> FullProfile(u, p) }
}

kotlin

// GET /logs/stream — SSE from MongoDB @Tailable cursor + heartbeat every 2s
@Bean
fun logStreamRoute(handler: LogStreamHandler): RouterFunction<ServerResponse> =
    RouterFunctions.route(GET("/logs/stream"), handler::stream)

// Handler: logRepo.findByStatus("ACTIVE") is @Tailable Flux
// Merge with Flux.interval(Duration.ofSeconds(2)).map { "" } for heartbeat
// MediaType.TEXT_EVENT_STREAM

Challenge 7: Anti-Pattern Hunt. Identify and fix: (A) .block() inside a stream, (B) Thread.sleep, (C) wrong scheduler. Refactor to a fully non-blocking pipeline.

kotlin

// Challenge 7: Anti-Pattern Hunt — fix the 3 reactive sins
// A: .block() — kills event loop
// B: Thread.sleep — blocks thread
// C: subscribeOn(immediate()) — wrong scheduler

fun processOrders(): Flux<OrderResponse> {
    return orderRepository.findAll()
        .flatMap { order ->
            userRepository.findById(order.userId)
                .map { user -> OrderResponse(order, user) }
        }
    // Remove .block(), Thread.sleep, use boundedElastic if you had blocking
}

Capstone: E-Commerce Checkout

Use Mono.justOrEmpty / switchIfEmpty for validation, flatMap and zipWith to combine balance and total, then map to the result string.

kotlin

// Capstone: E-Commerce Checkout
// userId + cartIds -> balance >= total ? "Approved" : "Rejected"
fun checkout(userId: String, cartIds: List<String>): Mono<String> {
    return Mono.justOrEmpty(userId).filter { it.isNotBlank() }
        .switchIfEmpty(Mono.error(IllegalArgumentException("Invalid User")))
        .flatMap { id -> getUserBalance(id) }
        .zipWith(Flux.fromIterable(cartIds).flatMap { getItemPrice(it) }.reduce(0.0, Double::plus))
        .map { (balance, total) -> if (balance >= total) "Approved" else "Rejected" }
}

Recap: Key Takeaways

Blocking one thread in WebFlux freezes a large part of your capacity; never block the event loop.
Nothing runs until you subscribe. Describe the pipeline, then subscribe.
map = synchronous (value → value). flatMap = async (value → Mono/Flux).
zip = parallel independent calls; flatMap = dependent async steps.
Use Schedulers.boundedElastic() for legacy blocking code; never block on the default scheduler.
Functional Router + Handler: testable, explicit WebFlux endpoints.
R2DBC (Postgres) and Reactive Mongo; avoid N+1 by using flatMap instead of map + .block().