| name | best-practices |
| description | Kotlin Coroutines with Spring Boot/WebFlux performance optimization and best practices guidelines |
| license | MIT |
| metadata | {"author":"devnogari","project_type":"kotlin-spring","rule_count":48,"categories":8} |
Kotlin Coroutines + Spring Boot Best Practices
Comprehensive performance optimization and best practices for Spring Boot applications with Kotlin Coroutines. 48 rules across 8 categories based on Spring Framework 7.0, Spring Boot 4.0, Kotlin 2.x coroutines, and R2DBC async patterns.
Activation Triggers
This skill activates when:
- Writing new controllers, services, or repositories with coroutines
- Implementing Flow patterns and reactive streams
- Optimizing async performance with WebFlux
- Reviewing or refactoring Kotlin coroutine code
- Debugging scope, cancellation, or flow issues
- Implementing authentication/authorization with reactive security
- Writing tests for coroutine-based applications
Rule Categories
| # | Category | Priority | Prefix | Rule Count | Focus |
|---|
| 1 | Coroutine Basics | CRITICAL | coroutine- | 6 | suspend functions, dispatchers |
| 2 | Scope Management | CRITICAL | scope- | 6 | GlobalScope avoidance, structured concurrency |
| 3 | Flow Patterns | HIGH | flow- | 6 | Flow, StateFlow, SharedFlow |
| 4 | R2DBC Database | HIGH | db- | 6 | Reactive repositories, transactions |
| 5 | Error Handling | HIGH | err- | 6 | SupervisorJob, exception handling |
| 6 | Spring WebFlux | MEDIUM | webflux- | 6 | Reactive endpoints, context propagation |
| 7 | Security | MEDIUM | sec- | 6 | Reactive security, JWT, context |
| 8 | Testing | MEDIUM | test- | 6 | runTest, MockK, Turbine |
Quick Reference - All 48 Rules
Coroutine Basics (CRITICAL) - coroutine-*
| Rule ID | Rule Name | Impact |
|---|
coroutine-suspend | Use suspend functions correctly | Non-blocking execution |
coroutine-dispatcher | Choose correct dispatcher | CPU/IO optimization |
coroutine-blocking | Never block in coroutines | Prevents thread starvation |
coroutine-context | Understand CoroutineContext | Proper context propagation |
coroutine-async | Use async/await for parallel | Concurrent execution |
coroutine-withContext | Use withContext for dispatcher switch | Clean context switching |
Scope Management (CRITICAL) - scope-*
| Rule ID | Rule Name | Impact |
|---|
scope-no-global | Never use GlobalScope | Prevents memory leaks |
scope-structured | Use structured concurrency | Automatic cleanup |
scope-coroutineScope | Use coroutineScope for grouping | Child coroutine management |
scope-supervisorScope | Use supervisorScope for isolation | Fault tolerance |
scope-lifecycle | Tie scope to lifecycle | Resource management |
scope-injection | Inject CoroutineScope | Testability |
Flow Patterns (HIGH) - flow-*
| Rule ID | Rule Name | Impact |
|---|
flow-cold | Understand Flow is cold | Proper subscription |
flow-operators | Use flow operators correctly | Efficient transformations |
flow-stateflow | Use StateFlow for state | UI state management |
flow-sharedflow | Use SharedFlow for events | Event broadcasting |
flow-flowOn | Use flowOn for context | Single context switch |
flow-collect | Collect safely in scope | Prevent leaks |
R2DBC Database (HIGH) - db-*
| Rule ID | Rule Name | Impact |
|---|
db-r2dbc | Use R2DBC not JDBC | Non-blocking DB |
db-repository | Use CoroutineCrudRepository | Coroutine-native API |
db-transaction | Handle transactions correctly | Data consistency |
db-connection-pool | Configure connection pool | Resource efficiency |
db-n-plus-one | Avoid N+1 queries | Query optimization |
db-migration | Use Flyway/Liquibase async | Schema management |
Error Handling (HIGH) - err-*
| Rule ID | Rule Name | Impact |
|---|
err-supervisor | Use SupervisorJob correctly | Fault isolation |
err-handler | Use CoroutineExceptionHandler | Centralized error handling |
err-try-catch | Catch exceptions in coroutines | Proper error recovery |
err-cancellation | Handle CancellationException | Correct cancellation |
err-flow-catch | Use catch operator in Flow | Flow error handling |
err-retry | Implement retry with backoff | Resilience |
Spring WebFlux (MEDIUM) - webflux-*
| Rule ID | Rule Name | Impact |
|---|
webflux-controller | Use suspend in controllers | Non-blocking endpoints |
webflux-coRouter | Use coRouter DSL | Coroutine-friendly routing |
webflux-webclient | Use WebClient with awaitBody | Async HTTP calls |
webflux-context | Propagate ReactorContext | Observability, tracing |
webflux-response | Return proper types | Efficient serialization |
webflux-filter | Use CoWebFilter | Coroutine-aware filters |
Security (MEDIUM) - sec-*
| Rule ID | Rule Name | Impact |
|---|
sec-reactive | Use ReactiveSecurityContextHolder | Reactive auth context |
sec-method | Use @PreAuthorize with suspend | Method-level security |
sec-context-propagation | Propagate SecurityContext | Cross-thread auth |
sec-jwt | Implement JWT properly | Stateless auth |
sec-cors | Configure CORS correctly | Cross-origin security |
sec-csrf | Handle CSRF in WebFlux | Request forgery protection |
Testing (MEDIUM) - test-*
| Rule ID | Rule Name | Impact |
|---|
test-runTest | Use runTest for coroutines | Proper test execution |
test-mockk | Use MockK coEvery/coVerify | Suspend function mocking |
test-turbine | Use Turbine for Flow testing | Flow assertions |
test-dispatcher | Use TestDispatcher | Deterministic tests |
test-webclient | Test with WebTestClient | Integration testing |
test-container | Use Testcontainers | Database testing |
Detailed Rules by Category
1. Coroutine Basics (CRITICAL)
coroutine-suspend - Use Suspend Functions Correctly
Impact: Non-blocking execution, proper coroutine integration
suspend fun getUser(id: Long): User {
Thread.sleep(1000)
return repository.findById(id)
}
suspend fun getUser(id: Long): User {
delay(1000)
return repository.findByIdOrNull(id)
?: throw UserNotFoundException(id)
}
suspend fun getUserWithOrders(userId: Long): UserWithOrders {
val user = userRepository.findById(userId)
val orders = orderRepository.findByUserId(userId)
return UserWithOrders(user, orders)
}
coroutine-dispatcher - Choose Correct Dispatcher
Impact: Optimized thread utilization for CPU vs IO operations
suspend fun processImage(data: ByteArray): ByteArray {
return withContext(Dispatchers.IO) {
heavyImageProcessing(data)
}
}
suspend fun processImage(data: ByteArray): ByteArray {
return withContext(Dispatchers.Default) {
heavyImageProcessing(data)
}
}
suspend fun readFile(path: String): String {
return withContext(Dispatchers.IO) {
File(path).readText()
}
}
suspend fun fetchFromApi(url: String): Response {
return webClient.get()
.uri(url)
.awaitExchange { it.awaitBody() }
}
coroutine-blocking - Never Block in Coroutines
Impact: Prevents thread starvation, maintains concurrency
@GetMapping("/data")
suspend fun getData(): Data {
val result = blockingHttpClient.get(url)
Thread.sleep(100)
return result
}
@GetMapping("/data")
suspend fun getData(): Data {
val result = webClient.get()
.uri(url)
.awaitBody<Data>()
delay(100)
return result
}
suspend fun legacyOperation(): Result {
return withContext(Dispatchers.IO) {
legacyBlockingClient.execute()
}
}
coroutine-context - Understand CoroutineContext
Impact: Proper context propagation for logging, tracing, auth
suspend fun processWithLogging() {
withContext(Dispatchers.Default) {
logger.info("Processing")
}
}
suspend fun processWithLogging() {
val mdcContext = MDCContext()
withContext(Dispatchers.Default + mdcContext) {
logger.info("Processing")
}
}
@GetMapping("/trace")
suspend fun tracedEndpoint(): Response {
return service.process()
}
coroutine-async - Use async/await for Parallel Execution
Impact: Concurrent execution, reduced latency
suspend fun getDashboard(userId: Long): Dashboard {
val user = userService.getUser(userId)
val orders = orderService.getOrders(userId)
val notifications = notificationService.get()
return Dashboard(user, orders, notifications)
}
suspend fun getDashboard(userId: Long): Dashboard = coroutineScope {
val user = async { userService.getUser(userId) }
val orders = async { orderService.getOrders(userId) }
val notifications = async { notificationService.get() }
Dashboard(
user = user.await(),
orders = orders.await(),
notifications = notifications.await()
)
}
suspend fun getDashboardSafe(userId: Long): Dashboard = supervisorScope {
val user = async { userService.getUser(userId) }
val orders = async {
runCatching { orderService.getOrders(userId) }
.getOrDefault(emptyList())
}
val notifications = async {
runCatching { notificationService.get() }
.getOrDefault(emptyList())
}
Dashboard(user.await(), orders.await(), notifications.await())
}
coroutine-withContext - Use withContext for Dispatcher Switch
Impact: Clean context switching without nested scopes
suspend fun processData(data: Data): Result {
return CoroutineScope(Dispatchers.Default).async {
heavyProcessing(data)
}.await()
}
suspend fun processData(data: Data): Result {
return withContext(Dispatchers.Default) {
heavyProcessing(data)
}
}
suspend fun processWithTimeout(data: Data): Result {
return withContext(Dispatchers.Default + CoroutineName("processor")) {
withTimeout(5000) {
heavyProcessing(data)
}
}
}
2. Scope Management (CRITICAL)
scope-no-global - Never Use GlobalScope
Impact: Prevents memory leaks, enables proper cancellation
@Service
class NotificationService {
fun sendAsync(notification: Notification) {
GlobalScope.launch {
sendNotification(notification)
}
}
}
@Service
class NotificationService(
private val scope: CoroutineScope
) {
fun sendAsync(notification: Notification) {
scope.launch {
sendNotification(notification)
}
}
}
@Configuration
class CoroutineConfig {
@Bean
fun applicationScope(): CoroutineScope {
return CoroutineScope(
SupervisorJob() +
Dispatchers.Default +
CoroutineName("app-scope")
)
}
@Bean
fun cleanupScope(scope: CoroutineScope): DisposableBean {
return DisposableBean { scope.cancel() }
}
}
scope-structured - Use Structured Concurrency
Impact: Automatic cleanup, proper parent-child relationships
suspend fun processOrders(orders: List<Order>) {
orders.forEach { order ->
CoroutineScope(Dispatchers.Default).launch {
processOrder(order)
}
}
}
suspend fun processOrders(orders: List<Order>) = coroutineScope {
orders.forEach { order ->
launch {
processOrder(order)
}
}
}
suspend fun processOrders(orders: List<Order>): List<Result> = coroutineScope {
orders.map { order ->
async {
processOrder(order)
}
}.awaitAll()
}
scope-coroutineScope - Use coroutineScope for Grouping
Impact: Child coroutine management, exception propagation
suspend fun updateUserProfile(userId: Long, profile: Profile): User = coroutineScope {
val validationDeferred = async { validateProfile(profile) }
val existingUser = async { userRepository.findById(userId) }
validationDeferred.await()
val user = existingUser.await() ?: throw UserNotFoundException(userId)
val updatedUser = user.copy(
name = profile.name,
email = profile.email
)
launch { auditService.logUpdate(userId) }
launch { cacheService.invalidate(userId) }
userRepository.save(updatedUser)
}
scope-supervisorScope - Use supervisorScope for Isolation
Impact: Fault tolerance, independent child failures
suspend fun sendNotifications(users: List<User>) = coroutineScope {
users.forEach { user ->
launch {
sendNotification(user)
}
}
}
suspend fun sendNotifications(users: List<User>) = supervisorScope {
users.forEach { user ->
launch {
try {
sendNotification(user)
} catch (e: Exception) {
logger.error("Failed to notify ${user.id}", e)
}
}
}
}
suspend fun fetchAllData(): AggregatedData = supervisorScope {
val users = async {
runCatching { userService.getAll() }.getOrDefault(emptyList())
}
val products = async {
runCatching { productService.getAll() }.getOrDefault(emptyList())
}
val orders = async {
runCatching { orderService.getAll() }.getOrDefault(emptyList())
}
AggregatedData(users.await(), products.await(), orders.await())
}
scope-lifecycle - Tie Scope to Lifecycle
Impact: Proper resource management, no orphaned coroutines
@Component
class BackgroundProcessor : DisposableBean {
private val scope = CoroutineScope(
SupervisorJob() + Dispatchers.Default
)
fun startProcessing() {
scope.launch {
while (isActive) {
processNextItem()
delay(1000)
}
}
}
override fun destroy() {
scope.cancel()
}
}
@RestController
class UserController {
@GetMapping("/users/{id}")
suspend fun getUser(@PathVariable id: Long): User {
return userService.getUser(id)
}
}
scope-injection - Inject CoroutineScope for Testability
Impact: Easy testing, flexible dispatcher configuration
@Service
class OrderProcessor(
private val orderRepository: OrderRepository,
private val scope: CoroutineScope = CoroutineScope(Dispatchers.Default)
) {
fun processAsync(orderId: Long): Job {
return scope.launch {
val order = orderRepository.findById(orderId)
processOrder(order)
}
}
}
@Test
fun `test order processing`() = runTest {
val testScope = this
val processor = OrderProcessor(mockRepository, testScope)
processor.processAsync(1L)
advanceUntilIdle()
coVerify { mockRepository.findById(1L) }
}
3. Flow Patterns (HIGH)
flow-cold - Understand Flow is Cold
Impact: Proper subscription behavior, no unexpected side effects
fun getUpdates(): Flow<Update> = flow {
println("Starting flow")
emit(fetchUpdate())
}
val updates = getUpdates()
suspend fun processUpdates() {
getUpdates().collect { update ->
println("Received: $update")
}
}
@Service
class EventService {
private val _events = MutableSharedFlow<Event>(
replay = 1,
extraBufferCapacity = 64
)
val events: SharedFlow<Event> = _events.asSharedFlow()
suspend fun emit(event: Event) {
_events.emit(event)
}
}
flow-operators - Use Flow Operators Correctly
Impact: Efficient transformations, proper backpressure handling
fun getProcessedOrders(): Flow<ProcessedOrder> {
return orderRepository.findAllAsFlow()
.filter { it.status == OrderStatus.PENDING }
.map { order -> processOrder(order) }
.catch { e ->
logger.error("Processing failed", e)
emit(ProcessedOrder.failed())
}
.onCompletion { logger.info("Processing complete") }
}
fun getNotifications(): Flow<Notification> {
return notificationSource.asFlow()
.buffer(capacity = 64)
.map { enrichNotification(it) }
}
fun getPriceUpdates(): Flow<Price> {
return priceSource.asFlow()
.conflate()
}
fun getSearchResults(queries: Flow<String>): Flow<List<Result>> {
return queries
.debounce(300)
.distinctUntilChanged()
.flatMapLatest { query ->
searchService.search(query)
}
}
flow-stateflow - Use StateFlow for State
Impact: Proper state management, UI updates
@Service
class ConnectionService {
private val _connectionState = MutableStateFlow(ConnectionState.DISCONNECTED)
val connectionState: StateFlow<ConnectionState> = _connectionState.asStateFlow()
suspend fun connect() {
_connectionState.value = ConnectionState.CONNECTING
try {
establishConnection()
_connectionState.value = ConnectionState.CONNECTED
} catch (e: Exception) {
_connectionState.value = ConnectionState.ERROR
}
}
}
class ViewModel {
private val _uiState = MutableStateFlow(UiState.Initial)
val uiState: StateFlow<UiState> = _uiState.asStateFlow()
}
flow-sharedflow - Use SharedFlow for Events
Impact: Event broadcasting, multiple subscribers
@Service
class EventBus {
private val _events = MutableSharedFlow<AppEvent>(
extraBufferCapacity = 64,
onBufferOverflow = BufferOverflow.DROP_OLDEST
)
val events: SharedFlow<AppEvent> = _events.asSharedFlow()
suspend fun publish(event: AppEvent) {
_events.emit(event)
}
fun tryPublish(event: AppEvent): Boolean {
return _events.tryEmit(event)
}
}
@Service
class ConfigService {
private val _config = MutableSharedFlow<Config>(
replay = 1
)
val config: SharedFlow<Config> = _config.asSharedFlow()
}
flow-flowOn - Use flowOn for Context Switching
Impact: Single context switch point, efficient execution
fun processItems(): Flow<Item> = flow {
items.forEach { item ->
val processed = withContext(Dispatchers.Default) {
process(item)
}
emit(processed)
}
}
fun processItems(): Flow<Item> = flow {
items.forEach { item ->
emit(process(item))
}
}.flowOn(Dispatchers.Default)
fun loadAndProcessData(): Flow<Result> {
return loadFromDb()
.flowOn(Dispatchers.IO)
.map { process(it) }
.flowOn(Dispatchers.Default)
}
flow-collect - Collect Safely in Scope
Impact: Prevent leaks, proper cancellation
@Service
class EventListener {
fun startListening() {
GlobalScope.launch {
events.collect { handleEvent(it) }
}
}
}
@Service
class EventListener(
private val scope: CoroutineScope
) : DisposableBean {
private var job: Job? = null
fun startListening() {
job = scope.launch {
events.collect { handleEvent(it) }
}
}
override fun destroy() {
job?.cancel()
}
}
@Service
class EventListener(
private val scope: CoroutineScope
) {
init {
events
.onEach { handleEvent(it) }
.launchIn(scope)
}
}
4. R2DBC Database (HIGH)
db-r2dbc - Use R2DBC Not JDBC
Impact: Non-blocking database operations
@Repository
class UserRepository(
private val jdbcTemplate: JdbcTemplate
) {
fun findById(id: Long): User? {
return jdbcTemplate.queryForObject(...)
}
}
@Repository
interface UserRepository : CoroutineCrudRepository<User, Long> {
suspend fun findByEmail(email: String): User?
@Query("SELECT * FROM users WHERE status = :status")
fun findByStatus(status: String): Flow<User>
}
@Configuration
@EnableR2dbcRepositories
class DatabaseConfig : AbstractR2dbcConfiguration() {
@Bean
override fun connectionFactory(): ConnectionFactory {
return ConnectionFactories.get(
ConnectionFactoryOptions.builder()
.option(DRIVER, "postgresql")
.option(HOST, "localhost")
.option(DATABASE, "mydb")
.option(USER, "user")
.option(PASSWORD, "pass")
.build()
)
}
}
db-repository - Use CoroutineCrudRepository
Impact: Native coroutine API, clean suspend functions
@Repository
interface OrderRepository : CoroutineCrudRepository<Order, Long> {
suspend fun findByOrderNumber(orderNumber: String): Order?
fun findByUserId(userId: Long): Flow<Order>
@Query("SELECT * FROM orders WHERE status = :status LIMIT :limit")
suspend fun findTopByStatus(status: String, limit: Int): List<Order>
}
@Service
class OrderService(
private val orderRepository: OrderRepository
) {
suspend fun getOrder(id: Long): Order {
return orderRepository.findById(id)
?: throw OrderNotFoundException(id)
}
fun getUserOrders(userId: Long): Flow<Order> {
return orderRepository.findByUserId(userId)
}
}
db-transaction - Handle Transactions Correctly
Impact: Data consistency, ACID compliance
suspend fun transferMoney(from: Long, to: Long, amount: BigDecimal) {
val fromAccount = accountRepository.findById(from)!!
fromAccount.balance -= amount
accountRepository.save(fromAccount)
val toAccount = accountRepository.findById(to)!!
toAccount.balance += amount
accountRepository.save(toAccount)
}
@Service
class TransferService(
private val transactionalOperator: TransactionalOperator,
private val accountRepository: AccountRepository
) {
suspend fun transferMoney(from: Long, to: Long, amount: BigDecimal) {
transactionalOperator.executeAndAwait {
val fromAccount = accountRepository.findById(from)
?: throw AccountNotFoundException(from)
val toAccount = accountRepository.findById(to)
?: throw AccountNotFoundException(to)
if (fromAccount.balance < amount) {
throw InsufficientFundsException()
}
accountRepository.save(fromAccount.copy(balance = fromAccount.balance - amount))
accountRepository.save(toAccount.copy(balance = toAccount.balance + amount))
}
}
}
@Service
class OrderService(
private val orderRepository: OrderRepository,
private val inventoryRepository: InventoryRepository
) {
@Transactional
suspend fun createOrder(order: Order): Order {
val savedOrder = orderRepository.save(order)
inventoryRepository.decrementStock(order.productId, order.quantity)
return savedOrder
}
}
db-connection-pool - Configure Connection Pool
Impact: Resource efficiency, connection reuse
@Configuration
class R2dbcConfig {
@Bean
fun connectionFactory(): ConnectionFactory {
val connectionFactory = ConnectionFactories.get(
ConnectionFactoryOptions.builder()
.option(DRIVER, "pool")
.option(PROTOCOL, "postgresql")
.option(HOST, "localhost")
.option(DATABASE, "mydb")
.option(USER, System.getenv("DB_USER"))
.option(PASSWORD, System.getenv("DB_PASSWORD"))
.build()
)
val poolConfig = ConnectionPoolConfiguration.builder(connectionFactory)
.maxIdleTime(Duration.ofMinutes(10))
.maxSize(20)
.initialSize(5)
.maxCreateConnectionTime(Duration.ofSeconds(5))
.validationQuery("SELECT 1")
.build()
return ConnectionPool(poolConfig)
}
}
# spring:
# r2dbc:
# url: r2dbc:pool:postgresql:
# pool:
# initial-size: 5
# max-size: 20
# max-idle-time: 10m
db-n-plus-one - Avoid N+1 Queries
Impact: Query optimization, reduced database load
suspend fun getUsersWithOrders(): List<UserWithOrders> {
val users = userRepository.findAll().toList()
return users.map { user ->
val orders = orderRepository.findByUserId(user.id).toList()
UserWithOrders(user, orders)
}
}
@Repository
interface UserRepository : CoroutineCrudRepository<User, Long> {
@Query("""
SELECT u.*, o.* FROM users u
LEFT JOIN orders o ON u.id = o.user_id
WHERE u.status = 'ACTIVE'
""")
fun findActiveUsersWithOrders(): Flow<UserWithOrders>
}
suspend fun getUsersWithOrders(): List<UserWithOrders> {
val users = userRepository.findAll().toList()
val userIds = users.map { it.id }
val ordersByUser = orderRepository.findByUserIdIn(userIds)
.toList()
.groupBy { it.userId }
return users.map { user ->
UserWithOrders(user, ordersByUser[user.id] ?: emptyList())
}
}
db-migration - Use Flyway/Liquibase
Impact: Schema version control, safe migrations
dependencies {
implementation("org.flywaydb:flyway-core")
runtimeOnly("org.postgresql:postgresql")
}
# spring:
# flyway:
# enabled: true
# locations: classpath:db/migration
# url: jdbc:postgresql:
# user: ${DB_USER}
# password: ${DB_PASSWORD}
# r2dbc:
# url: r2dbc:postgresql:
5. Error Handling (HIGH)
err-supervisor - Use SupervisorJob Correctly
Impact: Fault isolation, independent failure handling
val scope = CoroutineScope(SupervisorJob())
scope.launch {
throw Exception("Unhandled!")
}
val exceptionHandler = CoroutineExceptionHandler { _, exception ->
logger.error("Coroutine failed", exception)
errorReporter.report(exception)
}
val scope = CoroutineScope(
SupervisorJob() +
Dispatchers.Default +
exceptionHandler
)
scope.launch {
try {
riskyOperation()
} catch (e: Exception) {
logger.error("Operation failed", e)
}
}
err-handler - Use CoroutineExceptionHandler
Impact: Centralized error handling for uncaught exceptions
@Configuration
class CoroutineExceptionConfig {
@Bean
fun coroutineExceptionHandler(): CoroutineExceptionHandler {
return CoroutineExceptionHandler { context, exception ->
val name = context[CoroutineName]?.name ?: "unknown"
logger.error("Coroutine '$name' failed", exception)
when (exception) {
is CancellationException -> {
}
is NetworkException -> {
retryQueue.add(context)
}
else -> {
errorReporter.report(exception)
}
}
}
}
@Bean
fun applicationScope(handler: CoroutineExceptionHandler): CoroutineScope {
return CoroutineScope(SupervisorJob() + Dispatchers.Default + handler)
}
}
err-try-catch - Catch Exceptions in Coroutines
Impact: Proper error recovery, graceful degradation
suspend fun fetchData(): Data = coroutineScope {
val result = async {
riskyNetworkCall()
}
try {
result.await()
} catch (e: Exception) {
defaultData()
}
}
suspend fun fetchData(): Data = coroutineScope {
val result = async {
try {
riskyNetworkCall()
} catch (e: Exception) {
logger.warn("Network call failed, using default", e)
defaultData()
}
}
result.await()
}
suspend fun fetchData(): Data = supervisorScope {
val result = async {
riskyNetworkCall()
}
try {
result.await()
} catch (e: Exception) {
defaultData()
}
}
suspend fun fetchDataSafe(): Result<Data> {
return runCatching {
riskyNetworkCall()
}
}
err-cancellation - Handle CancellationException Properly
Impact: Correct cancellation propagation, resource cleanup
suspend fun processItems(items: List<Item>) {
items.forEach { item ->
try {
processItem(item)
} catch (e: Exception) {
logger.error("Failed", e)
}
}
}
suspend fun processItems(items: List<Item>) {
items.forEach { item ->
try {
processItem(item)
} catch (e: CancellationException) {
throw e
} catch (e: Exception) {
logger.error("Failed to process ${item.id}", e)
}
}
}
suspend fun processItems(items: List<Item>) {
items.forEach { item ->
runCatching {
processItem(item)
}.onFailure { e ->
if (e is CancellationException) throw e
logger.error("Failed to process ${item.id}", e)
}
}
}
suspend fun processWithResource() {
val resource = acquireResource()
try {
process(resource)
} finally {
withContext(NonCancellable) {
resource.close()
}
}
}
err-flow-catch - Use catch Operator in Flow
Impact: Flow error handling without breaking collection
fun getUpdates(): Flow<Update> = flow {
while (true) {
val update = fetchUpdate()
emit(update)
delay(1000)
}
}
fun getUpdates(): Flow<Update> = flow {
while (true) {
val update = fetchUpdate()
emit(update)
delay(1000)
}
}.catch { e ->
logger.error("Update fetch failed", e)
emit(Update.error(e.message))
}
fun getUpdates(): Flow<Update> = flow {
val update = fetchUpdate()
emit(update)
}.retry(3) { e ->
e is NetworkException
}.catch { e ->
emit(Update.error("Failed after retries"))
}
fun processNotifications(): Flow<Notification> {
return notificationSource
.onEach { notification ->
analyticsService.track(notification)
}
.catch { e ->
logger.error("Notification processing failed", e)
}
}
err-retry - Implement Retry with Backoff
Impact: Resilience for transient failures
suspend fun <T> retryWithBackoff(
maxRetries: Int = 3,
initialDelay: Long = 100,
maxDelay: Long = 1000,
factor: Double = 2.0,
block: suspend () -> T
): T {
var currentDelay = initialDelay
repeat(maxRetries - 1) { attempt ->
try {
return block()
} catch (e: Exception) {
if (e is CancellationException) throw e
logger.warn("Attempt ${attempt + 1} failed, retrying in ${currentDelay}ms", e)
}
delay(currentDelay)
currentDelay = (currentDelay * factor).toLong().coerceAtMost(maxDelay)
}
return block()
}
suspend fun fetchDataWithRetry(): Data {
return retryWithBackoff(maxRetries = 3) {
apiClient.fetchData()
}
}
fun getDataStream(): Flow<Data> = flow {
while (currentCoroutineContext().isActive) {
emit(fetchData())
delay(5000)
}
}.retryWhen { cause, attempt ->
if (cause is NetworkException && attempt < 3) {
delay(1000L * (attempt + 1))
true
} else {
false
}
}
6. Spring WebFlux (MEDIUM)
webflux-controller - Use Suspend in Controllers
Impact: Non-blocking endpoints, efficient resource use
@RestController
@RequestMapping("/api/users")
class UserController(
private val userService: UserService
) {
@GetMapping("/{id}")
suspend fun getUser(@PathVariable id: Long): User {
return userService.findById(id)
?: throw ResponseStatusException(HttpStatus.NOT_FOUND)
}
@GetMapping
fun getAllUsers(): Flow<User> {
return userService.findAll()
}
@PostMapping
@ResponseStatus(HttpStatus.CREATED)
suspend fun createUser(@RequestBody user: User): User {
return userService.save(user)
}
}
webflux-coRouter - Use coRouter DSL
Impact: Coroutine-friendly functional routing
@Configuration
class RouterConfig(
private val userHandler: UserHandler
) {
@Bean
fun userRoutes() = coRouter {
"/api/users".nest {
GET("", userHandler::getAllUsers)
GET("/{id}", userHandler::getUser)
POST("", userHandler::createUser)
PUT("/{id}", userHandler::updateUser)
DELETE("/{id}", userHandler::deleteUser)
}
filter { request, next ->
val start = System.currentTimeMillis()
val response = next(request)
val duration = System.currentTimeMillis() - start
logger.info("${request.method()} ${request.path()} - ${duration}ms")
response
}
}
}
@Component
class UserHandler(
private val userService: UserService
) {
suspend fun getUser(request: ServerRequest): ServerResponse {
val id = request.pathVariable("id").toLong()
val user = userService.findById(id)
return if (user != null) {
ServerResponse.ok().bodyValueAndAwait(user)
} else {
ServerResponse.notFound().buildAndAwait()
}
}
suspend fun getAllUsers(request: ServerRequest): ServerResponse {
val users = userService.findAll()
return ServerResponse.ok().bodyAndAwait(users)
}
}
webflux-webclient - Use WebClient with awaitBody
Impact: Non-blocking HTTP client calls
@Service
class ExternalApiService(
private val webClient: WebClient
) {
suspend fun fetchUser(userId: String): ExternalUser {
return webClient.get()
.uri("/users/{id}", userId)
.retrieve()
.awaitBody()
}
suspend fun fetchUserOrNull(userId: String): ExternalUser? {
return webClient.get()
.uri("/users/{id}", userId)
.retrieve()
.awaitBodyOrNull()
}
suspend fun createUser(user: ExternalUser): ExternalUser {
return webClient.post()
.uri("/users")
.bodyValue(user)
.retrieve()
.awaitBody()
}
fun fetchUsers(): Flow<ExternalUser> {
return webClient.get()
.uri("/users")
.retrieve()
.bodyToFlow()
}
}
@Configuration
class WebClientConfig {
@Bean
fun webClient(): WebClient {
return WebClient.builder()
.baseUrl("https://api.example.com")
.defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
.filter(ExchangeFilterFunction.ofRequestProcessor { request ->
logger.debug("Request: ${request.method()} ${request.url()}")
Mono.just(request)
})
.build()
}
}
webflux-context - Propagate ReactorContext
Impact: Observability, tracing, MDC propagation
@Configuration
class ContextPropagationConfig {
@Bean
fun contextPropagationHook(): Hooks.OnEachOperator {
return { publisher ->
if (publisher is Mono<*>) {
publisher.contextWrite { ctx ->
val mdc = MDC.getCopyOfContextMap() ?: emptyMap()
ctx.putAll(mdc.mapKeys { (k, _) -> k })
}
} else {
publisher
}
}
}
}
suspend fun <T> withMdcContext(block: suspend () -> T): T {
return withContext(MDCContext()) {
block()
}
}
@GetMapping("/traced")
suspend fun tracedEndpoint(): Response {
return withMdcContext {
MDC.put("requestId", UUID.randomUUID().toString())
logger.info("Processing request")
service.process()
}
}
webflux-response - Return Proper Types
Impact: Efficient serialization, proper HTTP semantics
@RestController
class ApiController(
private val itemService: ItemService
) {
@GetMapping("/items/{id}")
suspend fun getItem(@PathVariable id: Long): ResponseEntity<Item> {
val item = itemService.findById(id)
return if (item != null) {
ResponseEntity.ok(item)
} else {
ResponseEntity.notFound().build()
}
}
@GetMapping("/items", produces = [MediaType.APPLICATION_NDJSON_VALUE])
fun getItems(): Flow<Item> {
return itemService.findAll()
}
@GetMapping("/items/stream", produces = [MediaType.TEXT_EVENT_STREAM_VALUE])
fun streamItems(): Flow<ServerSentEvent<Item>> {
return itemService.findAll().map { item ->
ServerSentEvent.builder(item)
.id(item.id.toString())
.event("item")
.build()
}
}
@DeleteMapping("/items/{id}")
@ResponseStatus(HttpStatus.NO_CONTENT)
suspend fun deleteItem(@PathVariable id: Long) {
itemService.delete(id)
}
}
webflux-filter - Use CoWebFilter
Impact: Coroutine-aware request filtering
@Component
class LoggingFilter : CoWebFilter() {
override suspend fun filter(exchange: ServerWebExchange, chain: CoWebFilterChain) {
val request = exchange.request
val start = System.currentTimeMillis()
val requestId = UUID.randomUUID().toString()
exchange.attributes["requestId"] = requestId
try {
chain.filter(exchange)
} finally {
val duration = System.currentTimeMillis() - start
logger.info(
"[$requestId] ${request.method} ${request.path} - ${duration}ms"
)
}
}
}
@Component
@Order(Ordered.HIGHEST_PRECEDENCE)
class AuthenticationFilter(
private val tokenService: TokenService
) : CoWebFilter() {
override suspend fun filter(exchange: ServerWebExchange, chain: CoWebFilterChain) {
val token = exchange.request.headers.getFirst(HttpHeaders.AUTHORIZATION)
?.removePrefix("Bearer ")
if (token != null) {
val user = tokenService.validateAndGetUser(token)
if (user != null) {
exchange.attributes["currentUser"] = user
}
}
chain.filter(exchange)
}
}
7. Security (MEDIUM)
sec-reactive - Use ReactiveSecurityContextHolder
Impact: Reactive-compatible security context access
suspend fun getCurrentUser(): User {
val auth = SecurityContextHolder.getContext().authentication
return auth.principal as User
}
suspend fun getCurrentUser(): User {
return ReactiveSecurityContextHolder.getContext()
.awaitSingleOrNull()
?.authentication
?.principal as? User
?: throw UnauthorizedException()
}
suspend fun <T : Principal> ReactiveSecurityContextHolder.currentUserOrNull(): T? {
return getContext()
.awaitSingleOrNull()
?.authentication
?.principal as? T
}
@GetMapping("/me")
suspend fun getProfile(): UserProfile {
val user = ReactiveSecurityContextHolder.currentUserOrNull<User>()
?: throw ResponseStatusException(HttpStatus.UNAUTHORIZED)
return userService.getProfile(user.id)
}
sec-method - Use @PreAuthorize with Suspend
Impact: Method-level security on suspend functions
@Configuration
@EnableReactiveMethodSecurity
class SecurityConfig
@Service
class AdminService {
@PreAuthorize("hasRole('ADMIN')")
suspend fun deleteUser(userId: Long) {
userRepository.deleteById(userId)
}
@PreAuthorize("hasRole('ADMIN') or #userId == principal.id")
suspend fun updateUser(userId: Long, update: UserUpdate): User {
return userRepository.update(userId, update)
}
@PreAuthorize("@permissionService.canAccess(principal, #resourceId)")
suspend fun accessResource(resourceId: Long): Resource {
return resourceRepository.findById(resourceId)
}
}
sec-context-propagation - Propagate SecurityContext
Impact: Security context available across async boundaries
@Configuration
class SecurityContextConfig {
@Bean
fun securityContextCoroutineContext(): CoroutineContext {
return ReactorContext(
Context.of(SecurityContext::class.java,
ReactiveSecurityContextHolder.getContext())
)
}
}
class SecurityContextElement(
val securityContext: SecurityContext
) : AbstractCoroutineContextElement(Key) {
companion object Key : CoroutineContext.Key<SecurityContextElement>
}
suspend fun withSecurityContext(block: suspend () -> Unit) {
val securityContext = ReactiveSecurityContextHolder.getContext()
.awaitSingleOrNull()
if (securityContext != null) {
withContext(SecurityContextElement(securityContext)) {
block()
}
} else {
block()
}
}
sec-jwt - Implement JWT Properly
Impact: Stateless authentication, token validation
@Configuration
@EnableWebFluxSecurity
class SecurityConfig {
@Bean
fun securityWebFilterChain(http: ServerHttpSecurity): SecurityWebFilterChain {
return http
.csrf { it.disable() }
.httpBasic { it.disable() }
.formLogin { it.disable() }
.authorizeExchange { auth ->
auth.pathMatchers("/api/auth/**").permitAll()
auth.pathMatchers("/api/admin/**").hasRole("ADMIN")
auth.anyExchange().authenticated()
}
.oauth2ResourceServer { oauth2 ->
oauth2.jwt { jwt ->
jwt.jwtAuthenticationConverter(jwtAuthConverter())
}
}
.build()
}
@Bean
fun jwtAuthConverter(): ReactiveJwtAuthenticationConverter {
val converter = ReactiveJwtAuthenticationConverter()
converter.setJwtGrantedAuthoritiesConverter { jwt ->
val roles = jwt.getClaimAsStringList("roles") ?: emptyList()
Flux.fromIterable(roles.map { SimpleGrantedAuthority("ROLE_$it") })
}
return converter
}
}
@Service
class JwtTokenService(
@Value("\${jwt.secret}") private val secret: String,
@Value("\${jwt.expiration}") private val expiration: Long
) {
private val key = Keys.hmacShaKeyFor(secret.toByteArray())
fun generateToken(user: User): String {
val now = Date()
val expiryDate = Date(now.time + expiration)
return Jwts.builder()
.setSubject(user.id.toString())
.claim("email", user.email)
.claim("roles", user.roles)
.setIssuedAt(now)
.setExpiration(expiryDate)
.signWith(key, SignatureAlgorithm.HS256)
.compact()
}
suspend fun validateToken(token: String): User? {
return try {
val claims = Jwts.parserBuilder()
.setSigningKey(key)
.build()
.parseClaimsJws(token)
.body
val userId = claims.subject.toLong()
userRepository.findById(userId)
} catch (e: Exception) {
null
}
}
}
sec-cors - Configure CORS Correctly
Impact: Cross-origin security for SPAs
@Configuration
class CorsConfig {
@Bean
fun corsConfigurationSource(): CorsConfigurationSource {
val configuration = CorsConfiguration().apply {
allowedOrigins = listOf(
"https://myapp.com",
"https://staging.myapp.com"
)
allowedMethods = listOf("GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS")
allowedHeaders = listOf(
HttpHeaders.AUTHORIZATION,
HttpHeaders.CONTENT_TYPE,
"X-Request-ID"
)
exposedHeaders = listOf("X-Total-Count", "X-Page-Count")
allowCredentials = true
maxAge = 3600
}
return UrlBasedCorsConfigurationSource().apply {
registerCorsConfiguration("/api/**", configuration)
}
}
}
@Bean
fun securityWebFilterChain(http: ServerHttpSecurity): SecurityWebFilterChain {
return http
.cors { it.configurationSource(corsConfigurationSource()) }
.build()
}
sec-csrf - Handle CSRF in WebFlux
Impact: Request forgery protection
@Configuration
@EnableWebFluxSecurity
class SecurityConfig {
@Bean
fun securityWebFilterChain(http: ServerHttpSecurity): SecurityWebFilterChain {
return http
.csrf { it.disable() }
.build()
}
}
@Configuration
@EnableWebFluxSecurity
class StatefulSecurityConfig {
@Bean
fun securityWebFilterChain(http: ServerHttpSecurity): SecurityWebFilterChain {
return http
.csrf { csrf ->
csrf.csrfTokenRepository(
CookieServerCsrfTokenRepository.withHttpOnlyFalse()
)
}
.build()
}
}
8. Testing (MEDIUM)
test-runTest - Use runTest for Coroutines
Impact: Proper test execution, virtual time control
class UserServiceTest {
private val userRepository = mockk<UserRepository>()
private val userService = UserService(userRepository)
@Test
fun `should return user by id`() = runTest {
val user = User(1L, "test@example.com")
coEvery { userRepository.findById(1L) } returns user
val result = userService.getUser(1L)
assertEquals(user, result)
coVerify { userRepository.findById(1L) }
}
@Test
fun `should handle delay correctly`() = runTest {
coEvery { userRepository.findById(any()) } coAnswers {
delay(1000)
User(1L, "test@example.com")
}
val result = userService.getUser(1L)
assertNotNull(result)
}
}
test-mockk - Use MockK coEvery/coVerify
Impact: Suspend function mocking, clean test syntax
class OrderServiceTest {
private val orderRepository = mockk<OrderRepository>()
private val paymentService = mockk<PaymentService>()
private val orderService = OrderService(orderRepository, paymentService)
@Test
fun `should create order with payment`() = runTest {
val order = Order(productId = 1L, quantity = 2)
val savedOrder = order.copy(id = 1L)
coEvery { orderRepository.save(any()) } returns savedOrder
coEvery { paymentService.processPayment(any()) } returns PaymentResult.SUCCESS
val result = orderService.createOrder(order)
assertEquals(1L, result.id)
coVerify(exactly = 1) { orderRepository.save(order) }
coVerify(exactly = 1) { paymentService.processPayment(savedOrder) }
}
@Test
fun `should handle payment failure`() = runTest {
coEvery { orderRepository.save(any()) } returns Order(id = 1L)
coEvery { paymentService.processPayment(any()) } throws PaymentException("Failed")
assertThrows<PaymentException> {
orderService.createOrder(Order())
}
}
}
test-turbine - Use Turbine for Flow Testing
Impact: Clean Flow assertions, timeout handling
class EventServiceTest {
private val eventService = EventService()
@Test
fun `should emit events in order`() = runTest {
eventService.events.test {
eventService.emit(Event("first"))
eventService.emit(Event("second"))
assertEquals(Event("first"), awaitItem())
assertEquals(Event("second"), awaitItem())
cancelAndIgnoreRemainingEvents()
}
}
@Test
fun `should complete flow`() = runTest {
val flow = flowOf(1, 2, 3)
flow.test {
assertEquals(1, awaitItem())
assertEquals(2, awaitItem())
assertEquals(3, awaitItem())
awaitComplete()
}
}
@Test
fun `should handle flow errors`() = runTest {
val errorFlow = flow {
emit(1)
throw RuntimeException("Error")
}
errorFlow.test {
assertEquals(1, awaitItem())
val error = awaitError()
assertEquals("Error", error.message)
}
}
}
test-dispatcher - Use TestDispatcher
Impact: Deterministic tests, controlled execution
class TimerServiceTest {
@Test
fun `should emit at intervals`() = runTest {
val testDispatcher = StandardTestDispatcher(testScheduler)
val timerService = TimerService(testDispatcher)
val emissions = mutableListOf<Long>()
val job = launch {
timerService.ticker(1000).take(3).collect { emissions.add(it) }
}
advanceTimeBy(3000)
runCurrent()
assertEquals(3, emissions.size)
job.cancel()
}
@Test
fun `should use UnconfinedTestDispatcher for immediate execution`() = runTest {
val testDispatcher = UnconfinedTestDispatcher(testScheduler)
var executed = false
launch(testDispatcher) {
executed = true
}
assertTrue(executed)
}
}
class DataRefresher(
private val dispatcher: CoroutineDispatcher = Dispatchers.Default
) {
suspend fun refresh() = withContext(dispatcher) {
}
}
@Test
fun `should refresh data`() = runTest {
val refresher = DataRefresher(StandardTestDispatcher(testScheduler))
refresher.refresh()
advanceUntilIdle()
}
test-webclient - Test with WebTestClient
Impact: Integration testing for WebFlux endpoints
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
class UserControllerIntegrationTest {
@Autowired
lateinit var webTestClient: WebTestClient
@Test
fun `should get user by id`() {
webTestClient.get()
.uri("/api/users/1")
.exchange()
.expectStatus().isOk
.expectBody<User>()
.consumeWith { result ->
val user = result.responseBody!!
assertEquals(1L, user.id)
}
}
@Test
fun `should create user`() {
val newUser = UserCreate("test@example.com", "Test User")
webTestClient.post()
.uri("/api/users")
.bodyValue(newUser)
.exchange()
.expectStatus().isCreated
.expectBody<User>()
.consumeWith { result ->
val user = result.responseBody!!
assertNotNull(user.id)
assertEquals("test@example.com", user.email)
}
}
@Test
fun `should return 404 for missing user`() {
webTestClient.get()
.uri("/api/users/999")
.exchange()
.expectStatus().isNotFound
}
@Test
fun `should stream users`() {
webTestClient.get()
.uri("/api/users/stream")
.accept(MediaType.APPLICATION_NDJSON)
.exchange()
.expectStatus().isOk
.returnResult<User>()
.responseBody
.test()
.expectNextCount(10)
.verifyComplete()
}
}
test-container - Use Testcontainers
Impact: Real database testing, production-like tests
@SpringBootTest
@Testcontainers
class UserRepositoryIntegrationTest {
companion object {
@Container
val postgres = PostgreSQLContainer("postgres:15-alpine")
.withDatabaseName("testdb")
.withUsername("test")
.withPassword("test")
@JvmStatic
@DynamicPropertySource
fun properties(registry: DynamicPropertyRegistry) {
registry.add("spring.r2dbc.url") {
"r2dbc:postgresql://${postgres.host}:${postgres.getMappedPort(5432)}/${postgres.databaseName}"
}
registry.add("spring.r2dbc.username") { postgres.username }
registry.add("spring.r2dbc.password") { postgres.password }
registry.add("spring.flyway.url") { postgres.jdbcUrl }
registry.add("spring.flyway.user") { postgres.username }
registry.add("spring.flyway.password") { postgres.password }
}
}
@Autowired
lateinit var userRepository: UserRepository
@Test
fun `should save and retrieve user`() = runTest {
val user = User(email = "test@example.com", name = "Test User")
val saved = userRepository.save(user)
val retrieved = userRepository.findById(saved.id!!)
assertNotNull(retrieved)
assertEquals("test@example.com", retrieved?.email)
}
@Test
fun `should find users by email`() = runTest {
userRepository.save(User(email = "find@example.com", name = "Find Me"))
val found = userRepository.findByEmail("find@example.com")
assertNotNull(found)
assertEquals("Find Me", found?.name)
}
}
Project Structure
src/
├── main/kotlin/com/example/
│ ├── Application.kt # Main application
│ ├── config/
│ │ ├── CoroutineConfig.kt # Coroutine scope beans
│ │ ├── R2dbcConfig.kt # Database configuration
│ │ ├── SecurityConfig.kt # Security configuration
│ │ └── WebClientConfig.kt # HTTP client configuration
│ ├── controller/
│ │ └── UserController.kt # REST controllers
│ ├── domain/
│ │ └── User.kt # Domain entities
│ ├── repository/
│ │ └── UserRepository.kt # R2DBC repositories
│ ├── service/
│ │ └── UserService.kt # Business logic
│ └── exception/
│ └── Exceptions.kt # Custom exceptions
├── main/resources/
│ ├── application.yml # Configuration
│ └── db/migration/ # Flyway migrations
└── test/kotlin/com/example/
├── controller/
│ └── UserControllerTest.kt
├── service/
│ └── UserServiceTest.kt
└── repository/
└── UserRepositoryTest.kt
Integration Workflow
When implementing Kotlin Coroutine + Spring features:
- Before implementation: Review relevant rule categories
- During development: Apply patterns from quick reference
- Code review: Verify against rule checklist
- Performance issues: Consult detailed rules for solutions
coroutine-* rules → Proper suspend functions, no blocking
↓
scope-* rules → Structured concurrency, no GlobalScope
↓
flow-* rules → Correct Flow patterns and operators
↓
db-* rules → R2DBC, transactions, N+1 prevention
↓
err-* rules → SupervisorJob, proper exception handling
↓
webflux-* rules → Non-blocking endpoints, context propagation
↓
sec-* rules → Reactive security, JWT, context
↓
test-* rules → runTest, MockK, Turbine
Compliance Checklist
Before submitting Kotlin Coroutine + Spring code: