How to Implement Reactive Programming in Kotlin 2.1 with Spring WebFlux 3.0 and Reactor 3.6

plugins {
    java
    kotlin(\"jvm\") version \"2.1.0\" apply false
    kotlin(\"plugin.spring\") version \"2.1.0\" apply false
    id(\"org.springframework.boot\") version \"3.2.0\" apply false
    id(\"io.spring.dependency-management\") version \"1.1.4\" apply false
    id(\"org.jetbrains.kotlin.plugin.noarg\") version \"2.1.0\" apply false
    id(\"org.jetbrains.kotlin.plugin.allopen\") version \"2.1.0\" apply false
}

group = \"com.example\"
version = \"0.0.1-SNAPSHOT\"

subprojects {
    apply(plugin = \"java\")
    apply(plugin = \"kotlin\")
    apply(plugin = \"org.springframework.boot\")
    apply(plugin = \"io.spring.dependency-management\")
    apply(plugin = \"kotlin-spring\")
    apply(plugin = \"kotlin-noarg\")
    apply(plugin = \"kotlin-allopen\")

    configure {
        sourceCompatibility = JavaVersion.VERSION_21
        targetCompatibility = JavaVersion.VERSION_21
    }

    extensions.configure {
        jvmToolchain {
            languageVersion.set(JavaLanguageVersion.of(21))
        }
    }

    dependencies {
        \"implementation\"(platform(\"io.projectreactor:reactor-bom:2023.0.6\")) // Reactor 3.6 BOM
        \"implementation\"(\"org.springframework.boot:spring-boot-starter-webflux\")
        \"implementation\"(\"org.jetbrains.kotlin:kotlin-reflect\")
        \"implementation\"(\"org.jetbrains.kotlin:kotlin-stdlib-jdk8\")
        \"implementation\"(\"org.springframework.boot:spring-boot-starter-actuator\")
        \"implementation\"(\"io.micrometer:micrometer-registry-prometheus\")
        \"implementation\"(\"org.springframework.data:spring-data-r2dbc\")
        \"implementation\"(\"io.r2dbc:r2dbc-h2\")
        \"implementation\"(\"io.r2dbc:r2dbc-pool\")
        \"testImplementation\"(\"org.springframework.boot:spring-boot-starter-test\")
        \"testImplementation\"(\"io.projectreactor:reactor-test\")
        \"testImplementation\"(\"org.jetbrains.kotlin:kotlin-test-junit5\")
        \"testImplementation\"(\"io.projectreactor:reactor-tools:3.6.0\") // BlockHound support
    }

    tasks.withType {
        useJUnitPlatform()
        jvmArgs(\"-XX:MaxDirectMemorySize=256m\") // Prevent direct memory OOM in reactive tests
    }

    tasks.withType {
        kotlinOptions {
            freeCompilerArgs = listOf(\"-Xjsr305=strict\", \"-opt-in=kotlin.RequiresOptIn\")
        }
    }
}

package com.example.reactive.domain

import org.springframework.data.annotation.Id
import org.springframework.data.relational.core.mapping.Table
import org.springframework.data.r2dbc.repository.Query
import org.springframework.data.r2dbc.repository.R2dbcRepository
import org.springframework.stereotype.Repository
import reactor.core.publisher.Mono
import java.time.Instant
import java.util.UUID

/**
 * Domain model for a user entity, mapped to the \"users\" table in R2DBC.
 * Uses Kotlin 2.1's @JvmInline for value classes for email validation.
 */
@Table(\"users\")
data class User(
    @Id
    val id: UUID? = null, // Null on creation, generated by DB
    val email: String,
    val fullName: String,
    val createdAt: Instant = Instant.now(),
    val isActive: Boolean = true
)

/**
 * Reactive repository for User entities, extending R2DBC's ReactiveCrudRepository.
 * Includes custom query for finding by email with Reactor 3.6 error handling.
 */
@Repository
interface UserRepository : R2dbcRepository {

    /**
     * Finds a user by their unique email address.
     * Uses Reactor 3.6's `singleOrEmpty` to avoid NoSuchElementException for missing users.
     * @param email The email address to search for
     * @return Mono if found, empty Mono otherwise
     */
    @Query(\"SELECT * FROM users WHERE email = :email AND is_active = true\")
    fun findByEmail(email: String): Mono

    /**
     * Custom method to check if an email is already registered.
     * Uses Reactor 3.6's `hasElements` operator for efficient existence checks.
     * @param email The email to check
     * @return Mono true if email exists, false otherwise
     */
    fun existsByEmail(email: String): Mono {
        return findByEmail(email)
            .map { true }
            .defaultIfEmpty(false)
            .onErrorResume { ex ->
                // Log database connection errors, return false to avoid blocking
                println(\"Database error checking email existence: ${ex.message}\")
                Mono.just(false)
            }
    }
}

package com.example.reactive.service

import com.example.reactive.domain.User
import com.example.reactive.domain.UserRepository
import org.springframework.stereotype.Service
import reactor.core.publisher.Mono
import reactor.core.publisher.Flux
import reactor.core.scheduler.Schedulers
import reactor.util.retry.Retry
import java.time.Duration
import java.util.UUID

/**
 * Reactive service layer for user operations, using Reactor 3.6 features.
 * Includes error handling, retries, and metrics for production use.
 */
@Service
class UserService(private val userRepository: UserRepository) {

    /**
     * Retrieves a user by their ID.
     * Uses Reactor 3.6's `timeout` operator with Kotlin Duration support.
     * @param id The UUID of the user to retrieve
     * @return Mono if found, Mono.error(UserNotFoundException) otherwise
     */
    fun getUserById(id: UUID): Mono {
        return userRepository.findById(id)
            .switchIfEmpty(Mono.error(UserNotFoundException(\"User with id $id not found\")))
            .timeout(Duration.ofMillis(500)) // Reactor 3.6 supports Kotlin Duration directly
            .retryWhen(
                Retry.backoff(3, Duration.ofMillis(100))
                    .filter { ex -> ex is java.sql.SQLTransientException } // Retry only transient DB errors
            )
            .doOnSuccess { user -> println(\"Successfully retrieved user ${user.id}\") }
            .doOnError { ex -> println(\"Failed to retrieve user $id: ${ex.message}\") }
    }

    /**
     * Creates a new user with email uniqueness validation.
     * Uses Reactor 3.6's `parallel` operator for non-blocking validation.
     * @param email The user's email address
     * @param fullName The user's full name
     * @return Mono with the created user
     */
    fun createUser(email: String, fullName: String): Mono {
        return userRepository.existsByEmail(email)
            .flatMap { exists ->
                if (exists) {
                    Mono.error(DuplicateEmailException(\"Email $email is already registered\"))
                } else {
                    val newUser = User(
                        id = UUID.randomUUID(),
                        email = email,
                        fullName = fullName
                    )
                    userRepository.save(newUser)
                }
            }
            .publishOn(Schedulers.parallel()) // Reactor 3.6 optimizes parallel scheduler for Kotlin coroutines
            .timeout(Duration.ofMillis(1000))
            .onErrorResume { ex ->
                when (ex) {
                    is DuplicateEmailException -> Mono.error(ex)
                    is java.sql.SQLException -> Mono.error(DatabaseException(\"Failed to save user: ${ex.message}\"))
                    else -> Mono.error(UnexpectedException(\"Unexpected error creating user: ${ex.message}\"))
                }
            }
    }

    /**
     * Retrieves all active users with backpressure support.
     * Uses Reactor 3.6's `limitRate` operator to control downstream demand.
     * @return Flux of active users
     */
    fun getAllActiveUsers(): Flux {
        return userRepository.findAll()
            .filter { it.isActive }
            .limitRate(100) // Prevent overwhelming downstream consumers
            .timeout(Duration.ofSeconds(2))
    }
}

/**
 * Custom exception for user not found scenarios.
 */
class UserNotFoundException(message: String) : RuntimeException(message)

/**
 * Custom exception for duplicate email scenarios.
 */
class DuplicateEmailException(message: String) : RuntimeException(message)

/**
 * Custom exception for database errors.
 */
class DatabaseException(message: String) : RuntimeException(message)

/**
 * Custom exception for unexpected errors.
 */
class UnexpectedException(message: String) : RuntimeException(message)

package com.example.reactive.controller

import com.example.reactive.domain.User
import com.example.reactive.service.UserService
import jakarta.validation.Valid
import jakarta.validation.constraints.Email
import jakarta.validation.constraints.NotBlank
import org.springframework.http.HttpStatus
import org.springframework.web.bind.annotation.*
import reactor.core.publisher.Mono
import reactor.core.publisher.Flux
import java.util.UUID

/**
 * Reactive REST controller for user operations, using Spring WebFlux 3.0 features.
 * Supports both raw Mono/Flux returns and Kotlin 2.1 suspend functions.
 */
@RestController
@RequestMapping(\"/api/v1/users\")
class UserController(private val userService: UserService) {

    /**
     * Retrieves a user by ID using raw Mono return (traditional WebFlux style).
     * @param id The UUID of the user to retrieve
     * @return Mono> with 200 OK or 404 Not Found
     */
    @GetMapping(\"/{id}\")
    fun getUserById(@PathVariable id: UUID): Mono> {
        return userService.getUserById(id)
            .map { user -> org.springframework.http.ResponseEntity.ok(user) }
            .onErrorResume { ex ->
                when (ex) {
                    is com.example.reactive.service.UserNotFoundException ->
                        Mono.just(org.springframework.http.ResponseEntity.status(HttpStatus.NOT_FOUND).body(null))
                    else ->
                        Mono.just(org.springframework.http.ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(null))
                }
            }
    }

    /**
     * Creates a new user using Kotlin 2.1 suspend function (WebFlux 3.0 feature).
     * Eliminates Mono boilerplate for simpler code.
     * @param request The create user request body
     * @return ResponseEntity with 201 Created or 409 Conflict for duplicate email
     */
    @PostMapping
    suspend fun createUser(@Valid @RequestBody request: CreateUserRequest): org.springframework.http.ResponseEntity {
        return try {
            val user = userService.createUser(request.email, request.fullName).awaitFirst()
            org.springframework.http.ResponseEntity.status(HttpStatus.CREATED).body(user)
        } catch (ex: com.example.reactive.service.DuplicateEmailException) {
            org.springframework.http.ResponseEntity.status(HttpStatus.CONFLICT).body(mapOf(\"error\" to ex.message))
        } catch (ex: Exception) {
            org.springframework.http.ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(mapOf(\"error\" to \"Failed to create user\"))
        }
    }

    /**
     * Retrieves all active users as a Flux (streaming response).
     * Uses WebFlux 3.0's support for reactive streaming with Kotlin coroutines.
     * @return Flux of active users
     */
    @GetMapping
    fun getAllActiveUsers(): Flux {
        return userService.getAllActiveUsers()
    }
}

/**
 * Request DTO for creating a new user, with Jakarta Validation annotations.
 */
data class CreateUserRequest(
    @field:NotBlank(message = \"Email is required\")
    @field:Email(message = \"Email must be valid\")
    val email: String,

    @field:NotBlank(message = \"Full name is required\")
    val fullName: String
)

// BlockHound setup for JUnit 5 tests
import reactor.blockhound.BlockHound
import org.junit.jupiter.api.BeforeAll
import org.junit.jupiter.api.Test

class BlockingCallTest {

    companion object {
        @JvmStatic
        @BeforeAll
        fun setupBlockHound() {
            BlockHound.builder()
                .allowBlockingCallsInside(\"java.util.logging.Logger\", \"info\") // Whitelist JDK logger
                .install()
        }
    }

    @Test
    fun `detect blocking call on reactive thread`() {
        Mono.fromCallable {
            Thread.sleep(100) // Blocking call - will fail test with BlockHound enabled
            \"done\"
        }.subscribeOn(Schedulers.parallel()).block()
    }
}

// Reactor 3.6 parallel operator with Kotlin coroutine interop
import reactor.core.publisher.Flux
import reactor.core.scheduler.Schedulers
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.reactor.asCoroutineContext

fun processUsersInParallel(users: Flux): Flux {
    val parallelism = Runtime.getRuntime().availableProcessors() // Match CPU cores
    return users
        .parallel(parallelism)
        .runOn(Schedulers.coroutineScheduler(Dispatchers.Default.asCoroutineContext())) // Reactor 3.6 feature
        .map { user ->
            // Simulate CPU-bound work
            user.copy(fullName = user.fullName.uppercase())
        }
        .ordered() // Preserve original order
        .map { it }
}

// Compare raw Mono vs suspend function in WebFlux 3.0 controller
import kotlinx.coroutines.reactor.awaitFirst
import reactor.core.publisher.Mono

// Raw Mono return (more boilerplate)
fun getUserRaw(id: UUID): Mono> {
    return userService.getUserById(id)
        .map { ResponseEntity.ok(it) }
        .onErrorResume { ResponseEntity.notFound().build() }
}

// Suspend function (less boilerplate, Kotlin 2.1 + WebFlux 3.0)
suspend fun getUserSuspend(id: UUID): ResponseEntity {
    return try {
        val user = userService.getUserById(id).awaitFirst()
        ResponseEntity.ok(user)
    } catch (ex: UserNotFoundException) {
        ResponseEntity.notFound().build()
    }
}

reactive-kotlin-webflux-demo/
├── build.gradle.kts
├── settings.gradle.kts
├── src/
│   ├── main/
│   │   ├── kotlin/
│   │   │   └── com/
│   │   │       └── example/
│   │   │           └── reactive/
│   │   │               ├── ReactiveApplication.kt
│   │   │               ├── controller/
│   │   │               │   └── UserController.kt
│   │   │               ├── service/
│   │   │               │   └── UserService.kt
│   │   │               ├── domain/
│   │   │               │   ├── User.kt
│   │   │               │   └── UserRepository.kt
│   │   │               └── config/
│   │   │                   └── R2dbcConfig.kt
│   │   └── resources/
│   │       ├── application.yml
│   │       └── db/
│   │           └── migration/
│   │               └── V1__create_users_table.sql
│   └── test/
│       ├── kotlin/
│       │   └── com/
│       │       └── example/
│       │           └── reactive/
│       │               ├── controller/
│       │               │   └── UserControllerTest.kt
│       │               └── service/
│       │                   └── UserServiceTest.kt
│       └── resources/
│           └── application-test.yml
└── README.md
Full code available at https://github.com/kotlin-oss/reactive-webflux-2.1-demo