// Suspend функция - может вызывать другие suspend функции
suspend fun fetchUser(id: Long): User {
    delay(1000)  // приостанавливает выполнение на 1 секунду
    return userRepository.findById(id)
}

// Обычная функция НЕ может вызывать suspend функции напрямую
fun regularFunction() {
    // fetchUser(1)  // ОШИБКА - нужен coroutine scope
}

// Suspend функции можно вызывать только из:
// 1. Других suspend функций
// 2. Корутин (launch, async)
// 3. runBlocking (только для тестов/main)

// coroutineScope - все дочерние корутины связаны
suspend fun processData() = coroutineScope {
    val job1 = launch { task1() }
    val job2 = launch { task2() }
    // Если task1 упадет с ошибкой, task2 тоже будет отменена
}

// supervisorScope - дочерние корутины независимы
suspend fun processDataIndependently() = supervisorScope {
    val job1 = launch { task1() }
    val job2 = launch { task2() }
    // Если task1 упадет, task2 продолжит работу
}

// Практический пример: обработка списка независимых задач
suspend fun processUsers(users: List<User>) = supervisorScope {
    users.map { user ->
        launch {
            try {
                processUser(user)  // если обработка одного пользователя упадет
            } catch (e: Exception) {
                logger.error("Failed to process user ${user.id}", e)
            }
        }
    }.joinAll()  // дождаться завершения всех задач
}

// Простой launch
val job = GlobalScope.launch {
    println("Running in coroutine")
    delay(1000)
    println("Coroutine completed")
}

// Launch с обработкой ошибок
val job = CoroutineScope(Dispatchers.IO).launch {
    try {
        heavyOperation()
    } catch (e: Exception) {
        logger.error("Operation failed", e)
    }
}

job.cancel()  // отмена корутины
job.join()    // ожидание завершения

// Async для получения результата
suspend fun fetchUserData(userId: Long): UserData = coroutineScope {
    val userDeferred = async { userService.getUser(userId) }
    val profileDeferred = async { profileService.getProfile(userId) }
    
    // await() дожидается результата
    val user = userDeferred.await()
    val profile = profileDeferred.await()
    
    UserData(user, profile)
}

// Параллельное выполнение нескольких задач
suspend fun processMultipleUsers(ids: List<Long>): List<User> = coroutineScope {
    ids.map { id ->
        async { userService.getUser(id) }  // запускаем все параллельно
    }.awaitAll()  // дожидаемся всех результатов
}

suspend fun saveToDatabase(data: Data): Long = withContext(Dispatchers.IO) {
    // Переключаемся на IO поток для работы с БД
    database.save(data).id
}

suspend fun computeHash(data: ByteArray): String = withContext(Dispatchers.Default) {
    // Переключаемся на CPU-интенсивный поток
    MessageDigest.getInstance("SHA-256").digest(data).toHexString()
}

// Комбинирование контекстов
suspend fun processFile(file: File): ProcessResult {
    val content = withContext(Dispatchers.IO) {
        file.readText()  // IO операция
    }
    
    val processed = withContext(Dispatchers.Default) {
        heavyProcessing(content)  // CPU операция
    }
    
    return withContext(Dispatchers.IO) {
        saveResult(processed)  // снова IO операция
    }
}

// Dispatchers.Default - CPU-интенсивные задачи
launch(Dispatchers.Default) {
    val result = heavyComputation()  // вычисления, сортировка, парсинг
}

// Dispatchers.IO - блокирующие I/O операции
launch(Dispatchers.IO) {
    val data = database.query()     // БД запросы
    val response = httpClient.get() // HTTP запросы
    val content = file.readText()   // файловые операции
}

// Dispatchers.Unconfined - наследует поток вызывающего
launch(Dispatchers.Unconfined) {
    println("Thread: ${Thread.currentThread().name}")
    delay(100)
    println("Thread after delay: ${Thread.currentThread().name}")  // может измениться
}

// Dispatchers.Main - UI поток (Android/Desktop)
launch(Dispatchers.Main) {
    updateUI()  // обновление пользовательского интерфейса
}

// Ограниченный пул потоков для специфических задач
val dbDispatcher = Dispatchers.IO.limitedParallelism(5)

// Собственный Executor
val customDispatcher = Executors.newFixedThreadPool(10).asCoroutineDispatcher()

launch(dbDispatcher) {
    heavyDatabaseOperation()
}

// Не забывайте закрывать кастомные диспетчеры
customDispatcher.close()

// В coroutineScope исключение отменяет все дочерние корутины
suspend fun fragileOperation() = coroutineScope {
    launch { 
        delay(100)
        throw RuntimeException("Task 1 failed") 
    }
    launch { 
        delay(200)
        println("Task 2 completed")  // НЕ выполнится - будет отменена
    }
}

// В supervisorScope исключения изолированы
suspend fun resilientOperation() = supervisorScope {
    launch { 
        delay(100)
        throw RuntimeException("Task 1 failed") 
    }
    launch { 
        delay(200)
        println("Task 2 completed")  // выполнится независимо от Task 1
    }
}

// CoroutineExceptionHandler для необработанных исключений
val handler = CoroutineExceptionHandler { _, exception ->
    logger.error("Uncaught exception", exception)
}

// Применение на уровне scope
val scope = CoroutineScope(Dispatchers.Default + handler)
scope.launch {
    throw RuntimeException("This will be caught by handler")
}

// Try-catch в корутинах
suspend fun safeOperation() {
    try {
        riskyOperation()
    } catch (e: CustomException) {
        handleSpecificException(e)
    } catch (e: Exception) {
        logger.error("Unexpected error", e)
        throw e  // пробрасываем дальше если нужно
    }
}

class UserService {
    private val serviceScope = CoroutineScope(
        Dispatchers.Default + SupervisorJob() + 
        CoroutineExceptionHandler { _, e -> logger.error("Service error", e) }
    )
    
    suspend fun processUser(userId: Long) = serviceScope.async {
        // Все дочерние корутины будут отменены при отмене родительской
        val userData = fetchUserData(userId)
        val notifications = fetchNotifications(userId)
        
        UserProcessingResult(userData, notifications)
    }
    
    fun shutdown() {
        serviceScope.cancel()  // отменяет все активные корутины
    }
}

// Автоматическая отмена при выходе из scope
suspend fun processWithTimeout() = withTimeout(5000) {
    // Если операция займет больше 5 секунд, все дочерние корутины будут отменены
    coroutineScope {
        launch { longOperation1() }
        launch { longOperation2() }
    }
}

// Создание Flow
fun numbersFlow(): Flow<Int> = flow {
    for (i in 1..5) {
        delay(1000)
        emit(i)  // эмиссия значения
    }
}

// Операторы трансформации
fun processNumbers(): Flow<String> = numbersFlow()
    .filter { it % 2 == 0 }           // фильтрация
    .map { "Number: $it" }            // трансформация
    .flowOn(Dispatchers.Default)      // смена контекста

// Потребление Flow
suspend fun consumeFlow() {
    processNumbers().collect { value ->
        println(value)  // обработка каждого элемента
    }
}

// Flow для работы с БД
fun userUpdatesFlow(userId: Long): Flow<User> = flow {
    while (currentCoroutineContext().isActive) {
        emit(userRepository.findById(userId))
        delay(30000)  // обновление каждые 30 секунд
    }
}

// Создание канала
val channel = Channel<String>(capacity = Channel.BUFFERED)

// Producer корутина
launch {
    for (i in 1..5) {
        channel.send("Message $i")
        delay(1000)
    }
    channel.close()  // закрытие канала
}

// Consumer корутина
launch {
    for (message in channel) {  // итерация до закрытия
        println("Received: $message")
    }
}

// Типы каналов по емкости
val unlimited = Channel<Int>(Channel.UNLIMITED)    // без ограничений
val conflated = Channel<Int>(Channel.CONFLATED)    // только последнее значение
val rendezvous = Channel<Int>(Channel.RENDEZVOUS)  // синхронная передача

// SharedFlow для событий
class EventBus {
    private val _events = MutableSharedFlow<Event>(
        replay = 0,           // количество кешируемых событий
        extraBufferCapacity = 64  // дополнительный буфер
    )
    val events: SharedFlow<Event> = _events.asSharedFlow()
    
    fun publishEvent(event: Event) {
        _events.tryEmit(event)
    }
}

// StateFlow для состояния
class UserViewModel {
    private val _userState = MutableStateFlow<UserState>(UserState.Loading)
    val userState: StateFlow<UserState> = _userState.asStateFlow()
    
    suspend fun loadUser(id: Long) {
        _userState.value = UserState.Loading
        try {
            val user = userRepository.getUser(id)
            _userState.value = UserState.Success(user)
        } catch (e: Exception) {
            _userState.value = UserState.Error(e.message)
        }
    }
}

// Подписка на StateFlow
launch {
    userViewModel.userState.collect { state ->
        when (state) {
            is UserState.Loading -> showLoader()
            is UserState.Success -> displayUser(state.user)
            is UserState.Error -> showError(state.message)
        }
    }
}

// Выбор правильного инструмента:

// Flow - для трансформации данных
fun processFileFlow(file: File): Flow<ProcessedLine> = flow {
    file.useLines { lines ->
        lines.forEach { line ->
            emit(processLine(line))
        }
    }
}

// Channel - для producer-consumer паттерна
suspend fun processWithWorkers() {
    val workChannel = Channel<Work>(100)
    val resultChannel = Channel<Result>(100)
    
    // Workers
    repeat(5) {
        launch { worker(workChannel, resultChannel) }
    }
    
    // Producer
    launch { produceWork(workChannel) }
    
    // Results collector
    launch { collectResults(resultChannel) }
}

// SharedFlow - для событий системы
class NotificationService {
    private val _notifications = MutableSharedFlow<Notification>()
    val notifications = _notifications.asSharedFlow()
    
    fun sendNotification(notification: Notification) {
        _notifications.tryEmit(notification)
    }
}

// StateFlow - для состояния приложения
class AppState {
    private val _currentUser = MutableStateFlow<User?>(null)
    val currentUser = _currentUser.asStateFlow()
    
    fun setUser(user: User) {
        _currentUser.value = user
    }
}

// Простая корутина
suspend fun fetchUser(id: String): User {
    delay(1000) // приостанавливает корутину, НЕ блокирует поток
    return userRepository.findById(id)
}

// Потоки — дорого и ограниченно
repeat(100_000) {
    thread { 
        Thread.sleep(1000)
        println("Thread $it")
    }
} // OutOfMemoryError!

// Корутины — легко и масштабируемо
repeat(100_000) {
    GlobalScope.launch {
        delay(1000) // не блокирует поток
        println("Coroutine $it")
    }
} // Работает отлично!

// Thread — блокирующий подход
class ThreadExample {
    fun processRequests() {
        repeat(1000) { id ->
            thread {
                val result = blockingHttpCall(id) // блокирует поток
                processResult(result)
            }
        }
    }
}

// Coroutine — неблокирующий подход
class CoroutineExample {
    suspend fun processRequests() {
        repeat(1000) { id ->
            launch {
                val result = suspendingHttpCall(id) // не блокирует поток
                processResult(result)
            }
        }
    }
}

// ExecutorService — императивное управление потоками
class ExecutorExample {
    private val executor = Executors.newFixedThreadPool(10)
    
    fun processData(): CompletableFuture<String> {
        return CompletableFuture.supplyAsync({
            // сложная цепочка callbacks
            fetchDataFromApi()
                .thenCompose { data -> 
                    processData(data)
                }.thenApply { result ->
                    formatResult(result)
                }
        }, executor)
    }
    
    fun shutdown() {
        executor.shutdown() // ручное управление
    }
}

// CoroutineScope — декларативные корутины
class CoroutineExample : CoroutineScope {
    override val coroutineContext = Dispatchers.IO + SupervisorJob()
    
    suspend fun processData(): String {
        // линейный код, как синхронный
        val data = fetchDataFromApi()
        val processed = processData(data)
        return formatResult(processed)
    }
    
    fun cleanup() {
        coroutineContext.cancel() // автоматическая отмена всех дочерних корутин
    }
}

// suspend функция может быть приостановлена
suspend fun loadUserData(userId: String): UserData {
    val profile = loadProfile(userId) // может приостановиться
    val settings = loadSettings(userId) // может приостановиться
    return UserData(profile, settings)
}

// Обычная функция НЕ может вызывать suspend функции
fun regularFunction() {
    // loadUserData() // ОШИБКА! Нужен корутинный контекст
}

// launch — fire-and-forget, возвращает Job
val job = GlobalScope.launch {
    doSomething()
} // Unit

// async — возвращает Deferred<T> для получения результата
val deferred = GlobalScope.async {
    computeSomething()
} // Deferred<String>
val result = deferred.await()

// runBlocking — блокирует текущий поток (только для тестов/main)
runBlocking {
    val result = suspendingOperation()
    println(result)
}

// Main — UI поток (Android/Desktop)
launch(Dispatchers.Main) {
    updateUI(data)
}

// IO — операции ввода-вывода (сеть, файлы, БД)
launch(Dispatchers.IO) {
    val data = loadFromDatabase()
}

// Default — CPU-интенсивные задачи
launch(Dispatchers.Default) {
    val result = heavyComputation()
}

// Unconfined — продолжает в том же потоке (осторожно!)
launch(Dispatchers.Unconfined) {
    quickOperation()
}

class UserService : CoroutineScope {
    private val job = SupervisorJob()
    override val coroutineContext = Dispatchers.IO + job
    
    suspend fun loadUserDashboard(userId: String): Dashboard {
        // Все корутины автоматически отменятся при отмене родительской
        val profile = async { loadProfile(userId) }
        val notifications = async { loadNotifications(userId) }
        val recommendations = async { loadRecommendations(userId) }
        
        return Dashboard(
            profile = profile.await(),
            notifications = notifications.await(),
            recommendations = recommendations.await()
        )
    }
    
    fun shutdown() {
        job.cancel() // отменяет ВСЕ дочерние корутины
    }
}

// SupervisorJob — изолирует ошибки дочерних корутин
val supervisor = SupervisorJob()
val scope = CoroutineScope(Dispatchers.IO + supervisor)

scope.launch {
    try {
        val result1 = async { riskyOperation1() }
        val result2 = async { riskyOperation2() } // может упасть
        
        // Если result2 упадёт, result1 продолжит работу
        processResults(result1.await(), result2.await())
    } catch (e: Exception) {
        handleError(e)
    }
}

@RestController
class UserController(
    private val userService: UserService
) : CoroutineScope {
    
    override val coroutineContext = Dispatchers.IO + SupervisorJob()
    
    @GetMapping("/users/{id}/details")
    suspend fun getUserDetails(@PathVariable id: String): UserDetails {
        // Параллельное выполнение независимых операций
        val userDeferred = async { userService.findById(id) }
        val ordersDeferred = async { orderService.findByUserId(id) }
        val preferencesDeferred = async { preferencesService.findByUserId(id) }
        
        return UserDetails(
            user = userDeferred.await(),
            orders = ordersDeferred.await(),
            preferences = preferencesDeferred.await()
        )
    }
}

suspend fun processBatchOfUsers(userIds: List<String>) {
    userIds.chunked(10) // группы по 10
        .forEach { batch ->
            batch.map { userId ->
                async { processUser(userId) } // параллельно внутри группы
            }.awaitAll() // ждём завершения группы
        }
}

suspend fun fetchDataWithTimeout(): ApiResponse = withTimeout(5000) {
    val data = apiClient.fetchData() // автоматически отменится через 5 сек
    processData(data)
}

// Кооперативная отмена
suspend fun longRunningTask() {
    repeat(1000) { iteration ->
        ensureActive() // проверяет, не отменена ли корутина
        doSomeWork(iteration)
        delay(10) // suspension point — место для отмены
    }
}

// Thread — каждый поток имеет свой стек
Thread.start() // ~1-8 MB памяти

// Coroutine — объект в куче
launch { } // ~100 байт

// Blocking — поток заблокирован и не может делать другую работу
Thread.sleep(1000) // поток простаивает

// Suspension — поток освобождается для других корутин
delay(1000) // поток выполняет другие корутины

// Thread pool — ограниченное количество потоков
val executor = Executors.newFixedThreadPool(100) // максимум 100 одновременных задач

// Coroutines — практически неограниченная масштабируемость
repeat(1_000_000) { launch { } } // миллион корутин на нескольких потоках

// Callback hell
fetchUser(userId) { user ->
    fetchOrders(user.id) { orders ->
        fetchPayments(orders) { payments ->
            updateUI(user, orders, payments)
        }
    }
}

// Корутины — линейный код
val user = fetchUser(userId)
val orders = fetchOrders(user.id)
val payments = fetchPayments(orders)
updateUI(user, orders, payments)

// suspend функция может вызывать другие suspend функции
suspend fun fetchUserData(id: String): UserData {
    val profile = fetchProfile(id)    // suspension point
    val settings = fetchSettings(id)  // suspension point
    return UserData(profile, settings)
}

// Обычная функция НЕ может вызывать suspend функции напрямую
fun normalFunction() {
    // fetchUserData("123") // ОШИБКА компиляции!
}

suspend fun processData(): String {
    println("Начало обработки")
    delay(1000)              // suspension point - корутина приостанавливается
    println("Обработка завершена")
    return apiCall()         // suspension point - если apiCall() suspend
}

// GlobalScope — живёт всё время жизни приложения (осторожно!)
GlobalScope.launch { 
    // может пережить Activity/Fragment
}

// ViewModelScope — автоматически отменяется при очистке ViewModel
class UserViewModel : ViewModel() {
    fun loadData() {
        viewModelScope.launch {
            // автоматически отменится при onCleared()
        }
    }
}

// Кастомный scope с контролируемым жизненным циклом
class UserService : CoroutineScope {
    private val job = SupervisorJob()
    override val coroutineContext = Dispatchers.IO + job
    
    fun cleanup() {
        job.cancel() // отменяет ВСЕ дочерние корутины
    }
}

// launch возвращает Job для управления корутиной
val job: Job = scope.launch {
    repeat(1000) {
        doWork(it)
        delay(100)
    }
}

// Управление выполнением
job.cancel()           // отменить корутину
job.join()             // дождаться завершения
job.isActive           // проверить статус

// async возвращает Deferred<T> для получения результата
val userDeferred: Deferred<User> = scope.async { fetchUser(id) }
val ordersDeferred: Deferred<List<Order>> = scope.async { fetchOrders(id) }

// Получение результатов (параллельно загружались)
val user = userDeferred.await()
val orders = ordersDeferred.await()

// Блокирует текущий поток до завершения всех корутин
fun main() = runBlocking {
    val result = async { heavyComputation() }
    println("Результат: ${result.await()}")
}

// В тестах
@Test
fun testSuspendFunction() = runBlocking {
    val result = suspendingFunction()
    assertEquals(expected, result)
}

val job = launch {
    try {
        longRunningTask()
    } catch (e: CancellationException) {
        cleanup() // выполняется при отмене
        throw e   // важно перебросить
    }
}

// Управление Job
job.cancel()                    // отменить
job.cancelAndJoin()            // отменить и дождаться завершения
job.join()                     // дождаться завершения без отмены

// Состояния Job
job.isActive                   // выполняется ли
job.isCancelled               // отменена ли
job.isCompleted               // завершена ли

val deferred: Deferred<String> = async {
    expensiveComputation()
}

// Получение результата
val result = deferred.await()           // приостанавливает до получения результата
val resultOrNull = deferred.getCompleted() // не приостанавливает, может бросить исключение

// Deferred наследует Job
deferred.cancel()              // можно отменить
deferred.join()                // можно дождаться завершения

// Элементы контекста
val context = Dispatchers.IO +                    // диспетчер
              SupervisorJob() +                   // Job
              CoroutineName("UserService") +      // имя для отладки
              CoroutineExceptionHandler { _, ex -> // обработчик ошибок
                  log.error("Coroutine failed", ex)
              }

// Создание scope с контекстом
val scope = CoroutineScope(context)

class ApiService : CoroutineScope {
    override val coroutineContext = Dispatchers.IO + SupervisorJob()
    
    suspend fun fetchData(): Data {
        // Наследует IO dispatcher от родительского контекста
        val preliminaryData = loadFromCache()
        
        // Переключается на Default для CPU-интенсивной работы
        return withContext(Dispatchers.Default) {
            processData(preliminaryData)
        }
    }
}

suspend fun processUserData(userData: String): ProcessedData = 
    withContext(Dispatchers.Default) {
        // CPU-интенсивная обработка в Default dispatcher
        heavyDataProcessing(userData)
    } // автоматически возвращается в исходный контекст

suspend fun saveToDatabase(data: Data) {
    withContext(Dispatchers.IO) {
        // I/O операция в IO dispatcher
        database.save(data)
    }
}

// Main — основной поток (UI в Android/Desktop)
launch(Dispatchers.Main) {
    updateUI(data)        // обновление интерфейса
    progressBar.visible = false
}

// IO — операции ввода-вывода (сеть, файлы, БД)
launch(Dispatchers.IO) {
    val data = database.findAll()     // чтение из БД
    val response = apiClient.fetch()  // HTTP запрос
    writeToFile(data)                 // запись в файл
}

// Default — CPU-интенсивные задачи
launch(Dispatchers.Default) {
    val result = complexCalculation()  // математические вычисления
    val processed = sortLargeList()    // сортировка больших данных
    val compressed = compressData()    // сжатие данных
}

// Unconfined — продолжает в том же потоке (осторожно!)
launch(Dispatchers.Unconfined) {
    quickOperation()     // выполняется в текущем потоке
    delay(100)          // после suspension может оказаться в другом потоке!
}

class UserRepository {
    // Чтение из БД — IO dispatcher
    suspend fun findUser(id: String): User = withContext(Dispatchers.IO) {
        database.findById(id)
    }
    
    // Обработка данных — Default dispatcher  
    suspend fun processUserStats(users: List<User>): Statistics = 
        withContext(Dispatchers.Default) {
            users.groupBy { it.region }
                 .mapValues { it.value.size }
                 .let { Statistics(it) }
        }
}

class UserController {
    suspend fun getUserDashboard(id: String): UserDashboard {
        // Параллельные I/O операции
        val user = async(Dispatchers.IO) { userRepo.findUser(id) }
        val orders = async(Dispatchers.IO) { orderRepo.findByUserId(id) }
        val stats = async(Dispatchers.Default) { 
            calculateUserStats(id) // CPU-интенсивная задача
        }
        
        return UserDashboard(
            user = user.await(),
            orders = orders.await(), 
            stats = stats.await()
        )
    }
}

class OrderService : CoroutineScope {
    private val supervisorJob = SupervisorJob()
    override val coroutineContext = 
        Dispatchers.IO + supervisorJob + CoroutineName("OrderService")
    
    suspend fun processOrder(orderId: String): OrderResult = 
        withContext(coroutineContext) {
            try {
                // Параллельная загрузка данных
                val order = async { orderRepo.findById(orderId) }
                val user = async { userRepo.findById(order.await().userId) }
                val inventory = async(Dispatchers.IO) { 
                    inventoryService.checkAvailability(orderId) 
                }
                
                // CPU-интенсивная обработка
                val calculations = async(Dispatchers.Default) {
                    calculateOrderTotal(order.await(), inventory.await())
                }
                
                // Результат
                OrderResult(
                    order = order.await(),
                    user = user.await(),
                    total = calculations.await()
                )
            } catch (e: Exception) {
                handleOrderError(orderId, e)
                throw e
            }
        }
    
    fun shutdown() {
        supervisorJob.cancel() // отменяет все дочерние корутины
    }
}

// Structured concurrency в действии
suspend fun fetchUserDashboard(userId: String) = coroutineScope {
    // Все корутины автоматически отменятся, если функция будет отменена
    val profile = async { fetchProfile(userId) }
    val orders = async { fetchOrders(userId) }
    val recommendations = async { fetchRecommendations(userId) }
    
    Dashboard(
        profile = profile.await(),
        orders = orders.await(),
        recommendations = recommendations.await()
    )
    // Функция завершится только когда ВСЕ дочерние корутины завершатся
}

// ❌ Неструктурированный подход - проблемы
class BadUserService {
    fun loadUserData(userId: String) {
        GlobalScope.launch { fetchProfile(userId) }      // может "утечь"
        GlobalScope.launch { fetchOrders(userId) }       // никто не контролирует
        GlobalScope.launch { fetchNotifications(userId) } // нет связи между операциями
        
        // Проблемы:
        // 1. Невозможно отменить операции
        // 2. Нет гарантии завершения
        // 3. Утечки памяти
        // 4. Нет обработки ошибок
    }
}

// ✅ Структурированный подход - решение
class GoodUserService : CoroutineScope {
    private val job = SupervisorJob()
    override val coroutineContext = Dispatchers.IO + job
    
    suspend fun loadUserData(userId: String): UserData = coroutineScope {
        val profile = async { fetchProfile(userId) }
        val orders = async { fetchOrders(userId) }
        val notifications = async { fetchNotifications(userId) }
        
        UserData(
            profile = profile.await(),
            orders = orders.await(),
            notifications = notifications.await()
        )
    }
    
    fun cleanup() {
        job.cancel() // отменяет ВСЕ операции
    }
}

suspend fun processFiles(files: List<File>) = coroutineScope {
    files.map { file ->
        async {
            val data = readFile(file)        // открывает ресурс
            processData(data)
        }
    }.awaitAll()
    // Гарантия: все файлы будут закрыты, даже при отмене
}

suspend fun fetchAllData() = try {
    coroutineScope {
        val task1 = async { riskyOperation1() }
        val task2 = async { riskyOperation2() }  // может упасть
        val task3 = async { riskyOperation3() }
        
        // Если task2 упадёт, task1 и task3 автоматически отменятся
        Triple(task1.await(), task2.await(), task3.await())
    }
} catch (e: Exception) {
    handleError(e) // единая точка обработки ошибок
}

class DownloadManager : CoroutineScope {
    private val job = SupervisorJob()
    override val coroutineContext = Dispatchers.IO + job
    
    fun startDownloads(urls: List<String>) = launch {
        urls.forEach { url ->
            launch { downloadFile(url) } // все привязаны к родительскому scope
        }
    }
    
    fun cancelAllDownloads() {
        job.cancel() // одна команда отменяет все загрузки
    }
}

// Обычный Job - ошибка одного ребёнка отменяет всех
class RegularJobExample : CoroutineScope {
    override val coroutineContext = Dispatchers.IO + Job()
    
    fun processData() = launch {
        val task1 = launch { reliableOperation() }
        val task2 = launch { riskyOperation() }      // упадёт
        val task3 = launch { anotherOperation() }
        
        // Если task2 упадёт, task1 и task3 тоже отменятся!
    }
}

// SupervisorJob - изолирует ошибки детей
class SupervisorJobExample : CoroutineScope {
    override val coroutineContext = Dispatchers.IO + SupervisorJob()
    
    fun processData() = launch {
        val task1 = launch { reliableOperation() }
        val task2 = launch { riskyOperation() }      // упадёт
        val task3 = launch { anotherOperation() }
        
        // task2 упадёт, но task1 и task3 продолжат работу
    }
}

class ApiGateway : CoroutineScope {
    private val supervisorJob = SupervisorJob()
    override val coroutineContext = Dispatchers.IO + supervisorJob
    
    suspend fun fetchDashboardData(userId: String): DashboardData {
        // Каждый сервис независим - падение одного не влияет на другие
        val userService = async { 
            try { userServiceClient.getUser(userId) }
            catch (e: Exception) { null } // fallback
        }
        val orderService = async { 
            try { orderServiceClient.getOrders(userId) }
            catch (e: Exception) { emptyList() } // fallback
        }
        val notificationService = async { 
            try { notificationServiceClient.getNotifications(userId) }
            catch (e: Exception) { emptyList() } // fallback
        }
        
        return DashboardData(
            user = userService.await(),
            orders = orderService.await(),
            notifications = notificationService.await()
        )
    }
}

class BackgroundTaskManager : CoroutineScope {
    private val supervisorJob = SupervisorJob()
    override val coroutineContext = Dispatchers.Default + supervisorJob
    
    fun startBackgroundTasks() {
        // Каждая задача независима
        launch { 
            while (isActive) {
                cleanupOldData()
                delay(TimeUnit.HOURS.toMillis(1))
            }
        }
        
        launch {
            while (isActive) {
                generateReports()
                delay(TimeUnit.HOURS.toMillis(6))
            }
        }
        
        launch {
            while (isActive) {
                syncWithExternalSystem()
                delay(TimeUnit.MINUTES.toMillis(30))
            }
        }
    }
}

suspend fun fetchCriticalData() = coroutineScope {
    // Все операции критичны - если одна упадёт, отменяем всё
    val userData = async { fetchUser() }          // критично
    val accountData = async { fetchAccount() }    // критично  
    val securityData = async { fetchSecurity() } // критично
    
    CriticalData(
        user = userData.await(),
        account = accountData.await(),
        security = securityData.await()
    )
    // Если любая операция упадёт, все остальные отменятся
}

suspend fun fetchOptionalData() = supervisorScope {
    // Операции независимы - ошибка одной не влияет на другие
    val profile = async { 
        try { fetchProfile() }
        catch (e: Exception) { DefaultProfile }
    }
    val preferences = async { 
        try { fetchPreferences() }
        catch (e: Exception) { DefaultPreferences }
    }
    val recommendations = async { 
        try { fetchRecommendations() }
        catch (e: Exception) { emptyList() }
    }
    
    OptionalData(
        profile = profile.await(),
        preferences = preferences.await(),
        recommendations = recommendations.await()
    )
    // Каждая операция обрабатывает свои ошибки независимо
}

// Финансовая транзакция - всё или ничего
suspend fun processPayment(payment: Payment) = coroutineScope {
    val validation = async { validatePayment(payment) }
    val authorization = async { authorizePayment(payment) }
    val processing = async { processTransaction(payment) }
    
    // Если любой шаг упадёт, вся транзакция отменяется
    PaymentResult(
        validation.await(),
        authorization.await(), 
        processing.await()
    )
}

// Загрузка dashboard - показываем что получилось
suspend fun loadDashboard(userId: String) = supervisorScope {
    val weather = async { 
        try { weatherService.getCurrentWeather() }
        catch (e: Exception) { "Погода недоступна" }
    }
    val news = async { 
        try { newsService.getLatestNews() }
        catch (e: Exception) { emptyList() }
    }
    val stocks = async { 
        try { stockService.getPortfolio(userId) }
        catch (e: Exception) { emptyList() }
    }
    
    // Показываем то, что удалось загрузить
    Dashboard(weather.await(), news.await(), stocks.await())
}

class OrderProcessingService : CoroutineScope {
    private val supervisorJob = SupervisorJob()
    override val coroutineContext = Dispatchers.IO + supervisorJob
    
    suspend fun processOrder(order: Order): OrderResult = supervisorScope {
        // Критичная часть - обработка заказа
        val coreProcessing = async {
            coroutineScope { // строгая иерархия для критичных операций
                val validation = async { validateOrder(order) }
                val inventory = async { reserveInventory(order) }
                val payment = async { processPayment(order) }
                
                CoreOrderData(
                    validation.await(),
                    inventory.await(),
                    payment.await()
                )
            }
        }
        
        // Дополнительные сервисы - могут упасть без критичных последствий
        val notifications = async { 
            try { sendOrderNotification(order) }
            catch (e: Exception) { log.warn("Notification failed", e); null }
        }
        val analytics = async { 
            try { trackOrderEvent(order) }
            catch (e: Exception) { log.warn("Analytics failed", e); null }
        }
        val recommendations = async { 
            try { updateRecommendations(order) }
            catch (e: Exception) { log.warn("Recommendations failed", e); null }
        }
        
        OrderResult(
            core = coreProcessing.await(),
            notificationSent = notifications.await() != null,
            analyticsTracked = analytics.await() != null,
            recommendationsUpdated = recommendations.await() != null
        )
    }
}

class DataAggregationService {
    suspend fun aggregateUserData(userId: String): AggregatedData = 
        supervisorScope {
            val results = listOf(
                async { fetchFromSource1(userId) },
                async { fetchFromSource2(userId) },
                async { fetchFromSource3(userId) }
            ).map { deferred ->
                try {
                    deferred.await()
                } catch (e: Exception) {
                    log.warn("Data source failed", e)
                    null // fallback значение
                }
            }
            
            AggregatedData(
                source1Data = results[0],
                source2Data = results[1], 
                source3Data = results[2],
                completeness = results.count { it != null } / results.size.toDouble()
            )
        }
}

val job = launch {
    repeat(1000) { i ->
        println("Работаю: $i")
        delay(500) // suspension point - место для проверки отмены
    }
}

// Различные способы отмены
job.cancel()                           // простая отмена
job.cancel("Превышен таймаут")         // отмена с сообщением
job.cancelAndJoin()                    // отмена + ожидание завершения

// Отмена с причиной
job.cancel(TimeoutCancellationException("Операция слишком долгая"))

suspend fun longRunningTask() {
    repeat(1000) { iteration ->
        // Методы проверки отмены:
        
        ensureActive()           // бросает CancellationException если отменена
        
        if (!isActive) {         // мягкая проверка без исключения
            cleanup()
            return
        }
        
        yield()                  // проверяет отмену + даёт возможность другим корутинам
        
        doWork(iteration)
    }
}

class DataProcessor : CoroutineScope {
    private val job = SupervisorJob()
    override val coroutineContext = Dispatchers.Default + job
    
    fun processLargeDataset(data: List<Data>) = launch {
        data.forEachIndexed { index, item ->
            // Кооперативная проверка отмены
            if (!isActive) {
                saveProgress(index) // сохраняем прогресс
                return@launch      // корректный выход
            }
            
            processItem(item)
            
            // Альтернатива через ensureActive()
            ensureActive() // бросит CancellationException если отменена
        }
    }
}

suspend fun downloadFile(url: String, file: File) {
    var connection: HttpURLConnection? = null
    var outputStream: FileOutputStream? = null
    
    try {
        connection = createConnection(url)
        outputStream = FileOutputStream(file)
        
        val buffer = ByteArray(1024)
        while (isActive) { // проверяем отмену в цикле
            val bytesRead = connection.inputStream.read(buffer)
            if (bytesRead == -1) break
            
            outputStream.write(buffer, 0, bytesRead)
            yield() // suspension point для отмены
        }
    } catch (e: CancellationException) {
        file.delete() // очищаем частично скачанный файл
        throw e       // ВАЖНО: перебрасываем CancellationException
    } finally {
        connection?.disconnect()
        outputStream?.close()
    }
}

suspend fun handlingCancellation() {
    try {
        longRunningOperation()
    } catch (e: CancellationException) {
        // CancellationException НЕ считается ошибкой
        cleanup() // выполняем очистку
        throw e   // ОБЯЗАТЕЛЬНО перебрасываем!
    } catch (e: Exception) {
        // Обычные исключения обрабатываем
        handleError(e)
        throw e
    }
}

// CancellationException не ломает структуру
suspend fun parentOperation() = coroutineScope {
    try {
        val job1 = async { operation1() }
        val job2 = async { operation2() }
        
        job1.cancel() // отменяем job1
        
        // job2 продолжит работу, отмена job1 не влияет на него
        job2.await()
    } catch (e: CancellationException) {
        // Родительский scope получит CancellationException
        // только если отменён сам родитель
    }
}

suspend fun operationWithTimeout() {
    try {
        withTimeout(5000) { // автоматическая отмена через 5 секунд
            slowOperation()
        }
    } catch (e: TimeoutCancellationException) {
        // TimeoutCancellationException наследует CancellationException
        handleTimeout()
    }
}

// Мягкий timeout
suspend fun operationWithSoftTimeout() {
    val result = withTimeoutOrNull(5000) {
        slowOperation()
    }
    
    if (result == null) {
        handleTimeout()
    } else {
        processResult(result)
    }
}

fun launchExceptionPropagation() = runBlocking {
    val exceptionHandler = CoroutineExceptionHandler { _, exception ->
        println("Перехвачено в handler: ${exception.message}")
    }
    
    val scope = CoroutineScope(SupervisorJob() + exceptionHandler)
    
    scope.launch {
        throw RuntimeException("Ошибка в launch") // всплывает немедленно
    }
    
    scope.launch {
        delay(1000)
        println("Эта корутина продолжит работу") // выполнится
    }
    
    delay(2000)
}

fun asyncExceptionPropagation() = runBlocking {
    val scope = CoroutineScope(SupervisorJob())
    
    val deferred1 = scope.async {
        throw RuntimeException("Ошибка в async") // НЕ всплывает сразу
    }
    
    val deferred2 = scope.async {
        delay(1000)
        "Успешный результат"
    }
    
    try {
        // Исключение всплывёт здесь, при вызове await()
        val result1 = deferred1.await()
    } catch (e: Exception) {
        println("Перехвачено при await(): ${e.message}")
    }
    
    // deferred2 завершится успешно
    val result2 = deferred2.await()
    println("Результат: $result2")
}

suspend fun fetchMultipleResources(): CombinedResult = supervisorScope {
    val resource1 = async { 
        try { fetchResource1() }
        catch (e: Exception) { null } // локальная обработка
    }
    
    val resource2 = async { 
        try { fetchResource2() }
        catch (e: Exception) { null }
    }
    
    val resource3 = async { 
        try { fetchResource3() }
        catch (e: Exception) { null }
    }
    
    CombinedResult(
        resource1 = resource1.await(),
        resource2 = resource2.await(),
        resource3 = resource3.await()
    )
}

class RegularJobBehavior {
    fun demonstrateRegularJob() = runBlocking {
        val parentJob = Job()
        val scope = CoroutineScope(Dispatchers.Default + parentJob)
        
        val child1 = scope.launch {
            try {
                delay(2000)
                println("Child 1 завершён")
            } catch (e: CancellationException) {
                println("Child 1 отменён") // выполнится
            }
        }
        
        val child2 = scope.launch {
            delay(500)
            throw RuntimeException("Child 2 упал") // ломает всех
        }
        
        val child3 = scope.launch {
            try {
                delay(1000)
                println("Child 3 завершён") // НЕ выполнится
            } catch (e: CancellationException) {
                println("Child 3 отменён") // выполнится
            }
        }
        
        delay(3000)
        // Результат: child2 ломает child1 и child3
    }
}

class SupervisorJobBehavior {
    fun demonstrateSupervisorJob() = runBlocking {
        val supervisorJob = SupervisorJob()
        val scope = CoroutineScope(Dispatchers.Default + supervisorJob)
        
        val child1 = scope.launch {
            delay(2000)
            println("Child 1 завершён") // выполнится
        }
        
        val child2 = scope.launch {
            delay(500)
            throw RuntimeException("Child 2 упал") // изолированная ошибка
        }
        
        val child3 = scope.launch {
            delay(1000)
            println("Child 3 завершён") // выполнится
        }
        
        delay(3000)
        // Результат: только child2 упадёт, остальные работают
    }
}

class ApiService : CoroutineScope {
    private val supervisorJob = SupervisorJob()
    override val coroutineContext = Dispatchers.IO + supervisorJob + 
        CoroutineExceptionHandler { _, ex -> log.error("Unhandled exception", ex) }
    
    suspend fun loadDashboard(userId: String): Dashboard = supervisorScope {
        // Каждый компонент обрабатывает свои ошибки
        val profile = async {
            try { profileService.getProfile(userId) }
            catch (e: Exception) { 
                log.warn("Profile service failed", e)
                DefaultProfile(userId)
            }
        }
        
        val orders = async {
            try { orderService.getOrders(userId) }
            catch (e: Exception) {
                log.warn("Order service failed", e)
                emptyList<Order>()
            }
        }
        
        Dashboard(
            profile = profile.await(),
            orders = orders.await()
        )
    }
}

class ResilientApiClient {
    private var failureCount = 0
    private var lastFailureTime = 0L
    private val circuitBreakerTimeout = 30_000L // 30 секунд
    
    suspend fun makeRequest(request: ApiRequest): ApiResponse {
        // Проверяем circuit breaker
        if (isCircuitOpen()) {
            throw CircuitBreakerException("Service temporarily unavailable")
        }
        
        return try {
            val response = withTimeout(5000) {
                httpClient.execute(request)
            }
            resetFailureCount() // успех - сбрасываем счётчик
            response
        } catch (e: Exception) {
            incrementFailureCount()
            when (e) {
                is TimeoutCancellationException -> throw ServiceTimeoutException(e)
                is CancellationException -> throw e // не ломаем отмену
                else -> throw ServiceException("API call failed", e)
            }
        }
    }
    
    private fun isCircuitOpen(): Boolean {
        return failureCount >= 5 && 
               (System.currentTimeMillis() - lastFailureTime) < circuitBreakerTimeout
    }
}

suspend fun <T> withRetry(
    times: Int = 3,
    initialDelay: Long = 100,
    maxDelay: Long = 1000,
    factor: Double = 2.0,
    block: suspend () -> T
): T {
    var currentDelay = initialDelay
    
    repeat(times - 1) { attempt ->
        try {
            return block()
        } catch (e: CancellationException) {
            throw e // не ретраим отмену
        } catch (e: Exception) {
            log.warn("Attempt ${attempt + 1} failed", e)
            delay(currentDelay)
            currentDelay = (currentDelay * factor).toLong().coerceAtMost(maxDelay)
        }
    }
    
    return block() // последняя попытка без ретрая
}

// Использование
suspend fun reliableApiCall(): ApiResponse = withRetry(times = 3) {
    apiClient.fetchData()
}

class BatchProcessor {
    suspend fun processBatch(items: List<Item>): BatchResult = supervisorScope {
        val results = mutableListOf<ProcessingResult>()
        val activeJobs = mutableListOf<Job>()
        
        items.chunked(10).forEach { chunk ->
            val chunkJob = launch {
                chunk.forEach { item ->
                    if (isActive) { // проверяем отмену для каждого элемента
                        val result = processItem(item)
                        synchronized(results) { results.add(result) }
                    }
                }
            }
            activeJobs.add(chunkJob)
        }
        
        try {
            activeJobs.joinAll()
        } catch (e: CancellationException) {
            // Отменяем все активные задачи
            activeJobs.forEach { it.cancel() }
            throw e
        }
        
        BatchResult(results)
    }
}

class CompositeExceptionHandler {
    suspend fun processMultipleSources(): ProcessingResult = supervisorScope {
        val exceptions = mutableListOf<Exception>()
        val results = mutableListOf<SourceResult>()
        
        val jobs = dataSources.map { source ->
            async {
                try {
                    val result = source.fetchData()
                    synchronized(results) { results.add(result) }
                } catch (e: Exception) {
                    synchronized(exceptions) { exceptions.add(e) }
                }
            }
        }
        
        jobs.awaitAll()
        
        ProcessingResult(
            results = results,
            partialFailures = exceptions,
            success = exceptions.isEmpty()
        )
    }
}

// Базовая структура CoroutineContext
val context: CoroutineContext = 
    Dispatchers.IO +                    // где выполняется
    SupervisorJob() +                   // управление жизненным циклом  
    CoroutineName("ApiService") +       // имя для отладки
    CoroutineExceptionHandler { _, ex -> // обработка исключений
        log.error("Unhandled exception", ex)
    }

// Встроенные диспетчеры
val ioContext = Dispatchers.IO              // I/O операции (64+ потоков)
val defaultContext = Dispatchers.Default    // CPU задачи (CPU cores потоков)
val mainContext = Dispatchers.Main          // UI поток (1 поток)
val unconfinedContext = Dispatchers.Unconfined // текущий поток

// Кастомный диспетчер
val customDispatcher = Executors.newFixedThreadPool(4).asCoroutineDispatcher()
val customContext = customDispatcher + CoroutineName("CustomPool")

// Типы Job
val regularJob = Job()                    // обычный Job
val supervisorJob = SupervisorJob()       // изолирует ошибки детей

// Иерархия Job
val parentJob = SupervisorJob()
val childContext = Dispatchers.IO + Job(parent = parentJob)

// Управление Job
val job = launch(childContext) { 
    longRunningTask() 
}

job.cancel()        // отменить
job.join()          // дождаться завершения  
job.isActive        // проверить статус

// Именование корутин
val namedContext = CoroutineName("UserDataLoader") + Dispatchers.IO

launch(namedContext) {
    // В логах и профайлере будет видно имя "UserDataLoader"
    loadUserData()
}

// Получение имени из контекста
suspend fun getCurrentCoroutineName(): String? {
    return coroutineContext[CoroutineName]?.name
}

// Вложенные имена
val parentContext = CoroutineName("ApiService")
val childContext = parentContext + CoroutineName("DatabaseQuery")
// Результат: "ApiService.DatabaseQuery"

// Глобальный обработчик ошибок
val exceptionHandler = CoroutineExceptionHandler { context, exception ->
    val coroutineName = context[CoroutineName]?.name ?: "Unknown"
    log.error("Exception in coroutine '$coroutineName'", exception)
    
    // Можно отправить метрики, уведомления и т.д.
    metricsService.recordException(exception)
}

// Использование в контексте
val serviceContext = Dispatchers.IO + 
                    SupervisorJob() + 
                    CoroutineName("BackgroundService") +
                    exceptionHandler

class BackgroundService : CoroutineScope {
    override val coroutineContext = serviceContext
    
    fun startProcessing() = launch {
        // Любые необработанные исключения попадут в exceptionHandler
        processData()
    }
}

// Кастомный элемент контекста
class UserContext(val userId: String) : AbstractCoroutineContextElement(UserContext) {
    companion object Key : CoroutineContext.Key<UserContext>
    
    override fun toString(): String = "UserContext($userId)"
}

// Использование
val userContext = UserContext("user123") + Dispatchers.IO

launch(userContext) {
    val currentUser = coroutineContext[UserContext]?.userId
    println("Processing for user: $currentUser")
    
    withContext(Dispatchers.Default) {
        // UserContext автоматически передастся в новый поток
        val stillSameUser = coroutineContext[UserContext]?.userId
        println("Still processing for user: $stillSameUser")
    }
}

// Базовый контекст
val baseContext = Dispatchers.IO + SupervisorJob()

// Добавление элементов
val enrichedContext = baseContext + 
                     CoroutineName("ApiService") +
                     CoroutineExceptionHandler { _, ex -> handleError(ex) }

// Переопределение элементов (новый заменяет старый)
val newContext = enrichedContext + Dispatchers.Default  // заменяет Dispatchers.IO

suspend fun analyzeCurrentContext() {
    val context = coroutineContext
    
    // Получение элементов по ключу
    val dispatcher = context[ContinuationInterceptor] as? CoroutineDispatcher
    val job = context[Job]
    val name = context[CoroutineName]?.name
    
    // Проверка наличия элемента
    val hasExceptionHandler = context[CoroutineExceptionHandler] != null
    
    println("Dispatcher: $dispatcher")
    println("Job: $job")
    println("Name: $name")
    println("Has exception handler: $hasExceptionHandler")
}

// Удаление элемента из контекста
val contextWithoutName = originalContext.minusKey(CoroutineName)

// Фильтрация контекста
val filteredContext = originalContext.fold(EmptyCoroutineContext) { acc, element ->
    when (element) {
        is CoroutineName -> acc  // пропускаем CoroutineName
        else -> acc + element    // оставляем остальные элементы
    }
}

class ParentService : CoroutineScope {
    // Родительский контекст
    override val coroutineContext = Dispatchers.IO + 
                                   SupervisorJob() + 
                                   CoroutineName("ParentService")
    
    suspend fun processData() {
        // Дочерняя корутина наследует родительский контекст
        launch {
            // Автоматически получает: Dispatchers.IO + SupervisorJob + CoroutineName
            doWork()
        }
        
        // Переопределение элементов для дочерней корутины
        launch(Dispatchers.Default + CoroutineName("ChildWorker")) {
            // Получает: Dispatchers.Default + SupervisorJob + CoroutineName("ChildWorker")
            doCpuIntensiveWork()
        }
    }
}

suspend fun controlledInheritance() = coroutineScope {
    val parentContext = coroutineContext
    println("Parent context: $parentContext")
    
    // Полное наследование
    launch {
        println("Child 1 context: $coroutineContext") // идентичен родительскому
    }
    
    // Частичное переопределение
    launch(CoroutineName("CustomChild")) {
        println("Child 2 context: $coroutineContext") // изменено только имя
    }
    
    // Полное переопределение
    launch(Dispatchers.Unconfined + Job() + CoroutineName("IndependentChild")) {
        println("Child 3 context: $coroutineContext") // полностью новый контекст
    }
}

// Базовые контексты для разных типов сервисов
object ServiceContexts {
    // Для I/O операций
    val ioService = Dispatchers.IO + 
                   SupervisorJob() + 
                   CoroutineExceptionHandler { _, ex -> 
                       log.error("IO Service error", ex)
                   }
    
    // Для CPU-интенсивных задач
    val computeService = Dispatchers.Default + 
                        SupervisorJob() + 
                        CoroutineExceptionHandler { _, ex -> 
                            log.error("Compute Service error", ex)
                        }
    
    // Для UI операций (Android/Desktop)
    val uiService = Dispatchers.Main + 
                   SupervisorJob() + 
                   CoroutineExceptionHandler { _, ex -> 
                       showErrorDialog(ex)
                   }
}

// Использование
class UserRepository : CoroutineScope {
    override val coroutineContext = ServiceContexts.ioService + 
                                   CoroutineName("UserRepository")
}

// Контекст привязанный к HTTP запросу
class RequestContext(
    val requestId: String,
    val userId: String?,
    val traceId: String
) : AbstractCoroutineContextElement(RequestContext) {
    companion object Key : CoroutineContext.Key<RequestContext>
}

@RestController
class UserController {
    
    @GetMapping("/users/{id}")
    suspend fun getUser(@PathVariable id: String): UserDto = withContext(
        RequestContext(
            requestId = UUID.randomUUID().toString(),
            userId = id,
            traceId = MDC.get("traceId")
        )
    ) {
        // Весь код в этом блоке имеет доступ к RequestContext
        userService.findById(id)
    }
}

class UserService {
    suspend fun findById(id: String): UserDto {
        val requestContext = coroutineContext[RequestContext]
        log.info("Processing request ${requestContext?.requestId} for user $id")
        
        return userRepository.findById(id)
    }
}

class MetricsContext(
    val operationName: String,
    private val startTime: Long = System.currentTimeMillis()
) : AbstractCoroutineContextElement(MetricsContext) {
    companion object Key : CoroutineContext.Key<MetricsContext>
    
    fun recordDuration() {
        val duration = System.currentTimeMillis() - startTime
        metricsRegistry.timer(operationName).record(duration, TimeUnit.MILLISECONDS)
    }
}

// Автоматическое измерение времени выполнения
suspend fun <T> withMetrics(operationName: String, block: suspend () -> T): T {
    val metricsContext = MetricsContext(operationName)
    return try {
        withContext(metricsContext) {
            block()
        }
    } finally {
        metricsContext.recordDuration()
    }
}

// Использование
suspend fun processOrder(orderId: String) = withMetrics("order.processing") {
    // Время выполнения автоматически измеряется
    orderRepository.findById(orderId)
        .let { validateOrder(it) }
        .let { processPayment(it) }
        .let { updateInventory(it) }
}

class AdaptiveService : CoroutineScope {
    private var _context = Dispatchers.IO + SupervisorJob()
    override val coroutineContext: CoroutineContext get() = _context
    
    fun adaptToLoad(currentLoad: Int) {
        val newDispatcher = when {
            currentLoad > 80 -> Dispatchers.IO.limitedParallelism(32)
            currentLoad > 50 -> Dispatchers.IO.limitedParallelism(16)
            else -> Dispatchers.IO
        }
        
        _context = newDispatcher + 
                  _context[Job]!! + 
                  CoroutineName("AdaptiveService-Load$currentLoad")
    }
}

suspend fun smartOperation() {
    val currentDispatcher = coroutineContext[ContinuationInterceptor]
    
    when (currentDispatcher) {
        Dispatchers.Main -> {
            // Уже в UI потоке - можно обновлять UI
            updateProgressBar()
        }
        Dispatchers.IO -> {
            // В IO потоке - можно делать блокирующие операции
            performBlockingIO()
        }
        else -> {
            // Переключаемся в подходящий контекст
            withContext(Dispatchers.IO) {
                performBlockingIO()
            }
        }
    }
}

// Базовый пример Flow
val numbersFlow = flow {
    for (i in 1..5) {
        delay(1000)    // асинхронная операция
        emit(i)        // излучение значения
    }
}

// Использование
numbersFlow.collect { value ->
    println("Received: $value") // получение каждого значения
}

val coldFlow = flow {
    println("Starting cold flow") // выполнится для каждого collect
    repeat(3) { i ->
        delay(1000)
        emit(i)
    }
}

// Каждый collect запускает flow заново
launch { coldFlow.collect { println("Collector 1: $it") } }
launch { coldFlow.collect { println("Collector 2: $it") } }
// Результат: "Starting cold flow" напечатается дважды

// Cold flow как factory функция
fun createDataStream(userId: String) = flow {
    val data = fetchUserData(userId) // выполняется при каждом collect
    emit(data)
}

// SharedFlow - hot stream
val hotFlow = MutableSharedFlow<Int>()

// Запуск производителя данных
launch {
    repeat(5) { i ->
        delay(1000)
        hotFlow.emit(i) // излучает независимо от подписчиков
    }
}

// Подписчики получают только новые данные
launch {
    delay(2500) // подписываемся через 2.5 секунды
    hotFlow.collect { println("Late subscriber: $it") } // получит только 3, 4
}

val numbers = flowOf(1, 2, 3, 4, 5)

val doubled = numbers.map { it * 2 }
// Результат: 2, 4, 6, 8, 10

val userNames = userIds.map { id ->
    userRepository.findById(id).name // suspend функция
}

val evenNumbers = numbers.filter { it % 2 == 0 }
// Результат: 2, 4

val activeUsers = users.filter { user ->
    userService.isActive(user.id) // suspend функция
}

val userOrders = userIds.flatMapConcat { userId ->
    orderRepository.getOrdersFlow(userId) // возвращает Flow<Order>
}
// Ждёт завершения каждого потока перед переходом к следующему

// Практический пример
fun getUserDataFlow(userIds: List<String>) = userIds.asFlow()
    .flatMapConcat { userId ->
        flow {
            val profile = userService.getProfile(userId)
            val orders = orderService.getOrders(userId)
            emit(UserData(profile, orders))
        }
    }

// Простая подписка
flow.collect { value ->
    println("Received: $value")
}

// Обработка с сохранением в список
val results = mutableListOf<String>()
flow.collect { results.add(it) }

// Collect с обработкой ошибок
try {
    flow.collect { processData(it) }
} catch (e: Exception) {
    handleError(e)
}

val processedFlow = dataFlow
    .onEach { data -> 
        log.info("Processing data: $data") // side effect
    }
    .onEach { data ->
        metricsService.recordDataPoint(data) // ещё один side effect
    }
    .map { processData(it) } // основная трансформация

// onEach не изменяет сам поток данных
processedFlow.collect { result ->
    // получаем обработанные данные
}

val resilientFlow = riskyDataFlow
    .map { data -> 
        riskyTransformation(data) // может бросить исключение
    }
    .catch { exception ->
        when (exception) {
            is NetworkException -> emit(CachedData) // fallback значение
            is ValidationException -> { 
                log.warn("Validation failed", exception)
                // не emit - пропускаем невалидные данные
            }
            else -> throw exception // перебрасываем неожиданные ошибки
        }
    }

val optimizedFlow = databaseFlow
    .flowOn(Dispatchers.IO)        // upstream операции в IO
    .map { data -> 
        heavyComputation(data)     // CPU-интенсивная обработка
    }
    .flowOn(Dispatchers.Default)   // эта операция в Default
    .onEach { result ->
        updateUI(result)           // UI операции остаются в текущем контексте
    }

// Практический пример
class DataProcessor {
    fun processDataStream() = flow {
        // Эмиссия в IO контексте
        val rawData = loadFromDatabase()
        emit(rawData)
    }
    .flowOn(Dispatchers.IO)
    .map { data ->
        // Обработка в Default контексте  
        processData(data)
    }
    .flowOn(Dispatchers.Default)
}

class UserViewModel {
    private val _userState = MutableStateFlow<UiState>(UiState.Loading)
    val userState: StateFlow<UiState> = _userState.asStateFlow()
    
    suspend fun loadUser(id: String) {
        _userState.value = UiState.Loading
        try {
            val user = userRepository.findById(id)
            _userState.value = UiState.Success(user)
        } catch (e: Exception) {
            _userState.value = UiState.Error(e.message)
        }
    }
}

// Подписка
userViewModel.userState.collect { state ->
    when (state) {
        is UiState.Loading -> showProgress()
        is UiState.Success -> showUser(state.user)
        is UiState.Error -> showError(state.message)
    }
}

class EventBus {
    private val _events = MutableSharedFlow<AppEvent>()
    val events: SharedFlow<AppEvent> = _events.asSharedFlow()
    
    suspend fun publishEvent(event: AppEvent) {
        _events.emit(event)
    }
}

// Разные типы событий
sealed class AppEvent {
    data class UserLoggedIn(val userId: String) : AppEvent()
    data class OrderCreated(val orderId: String) : AppEvent()
    object NetworkError : AppEvent()
}

// Подписка на события
eventBus.events.collect { event ->
    when (event) {
        is AppEvent.UserLoggedIn -> handleUserLogin(event.userId)
        is AppEvent.OrderCreated -> sendNotification(event.orderId)
        is AppEvent.NetworkError -> showNetworkError()
    }
}

// MutableStateFlow - простое состояние
val simpleState = MutableStateFlow("initial")

// MutableSharedFlow с настройками
val configuredSharedFlow = MutableSharedFlow<String>(
    replay = 2,           // последние 2 значения для новых подписчиков
    extraBufferCapacity = 10, // дополнительный буфер
    onBufferOverflow = BufferOverflow.DROP_OLDEST // стратегия при переполнении
)

// Сложные конфигурации для разных сценариев
class NotificationService {
    // Критичные уведомления - не теряем
    private val criticalEvents = MutableSharedFlow<CriticalEvent>(
        replay = 0,
        extraBufferCapacity = Int.MAX_VALUE,
        onBufferOverflow = BufferOverflow.SUSPEND
    )
    
    // Обычные события - можем потерять старые
    private val regularEvents = MutableSharedFlow<RegularEvent>(
        replay = 1,
        extraBufferCapacity = 64,
        onBufferOverflow = BufferOverflow.DROP_OLDEST
    )
}

// LiveData подход
class UserViewModelLiveData : ViewModel() {
    private val _users = MutableLiveData<List<User>>()
    val users: LiveData<List<User>> = _users
    
    fun loadUsers() {
        viewModelScope.launch {
            _users.value = userRepository.getUsers()
        }
    }
}

// Flow подход
class UserViewModelFlow : ViewModel() {
    val users: StateFlow<List<User>> = userRepository.getUsersFlow()
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5000),
            initialValue = emptyList()
        )
}

// Подписка в UI (Android)
class UserFragment : Fragment() {
    override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
        // LiveData подписка
        viewModel.users.observe(viewLifecycleOwner) { users ->
            updateUI(users)
        }
        
        // Flow подписка
        viewLifecycleOwner.lifecycleScope.launch {
            viewModel.users.collect { users ->
                updateUI(users)
            }
        }
    }
}

class DataService : CoroutineScope {
    private val job = SupervisorJob()
    override val coroutineContext = Dispatchers.IO + job
    
    fun startListening() {
        // Подписка привязана к scope службы
        launch {
            dataFlow.collect { data ->
                processData(data)
            }
        }
    }
    
    fun stop() {
        job.cancel() // автоматически отменяет все подписки
    }
}

suspend fun conditionalCollection() {
    dataFlow
        .takeWhile { data -> data.isValid } // автоматическая отмена при условии
        .collect { data ->
            processValidData(data)
        }
    // collect завершится когда встретится невалидный элемент
}

// Отмена по времени
suspend fun timedCollection() = withTimeout(30_000) {
    dataFlow.collect { data ->
        processData(data)
    }
    // автоматическая отмена через 30 секунд
}

fun createCancellableFlow() = flow {
    try {
        repeat(100) { i ->
            ensureActive() // проверка отмены
            val data = fetchData(i)
            emit(data)
            delay(1000)
        }
    } catch (e: CancellationException) {
        cleanup() // очистка ресурсов при отмене
        throw e   // важно перебросить
    }
}

class UserRepository {
    private val _users = MutableStateFlow<List<User>>(emptyList())
    val users: StateFlow<List<User>> = _users.asStateFlow()
    
    private val _events = MutableSharedFlow<UserEvent>()
    val events: SharedFlow<UserEvent> = _events.asSharedFlow()
    
    suspend fun addUser(user: User) {
        val updatedUsers = _users.value + user
        _users.value = updatedUsers
        _events.emit(UserEvent.UserAdded(user))
    }
    
    suspend fun removeUser(userId: String) {
        val updatedUsers = _users.value.filterNot { it.id == userId }
        _users.value = updatedUsers
        _events.emit(UserEvent.UserRemoved(userId))
    }
}

class StockPriceService {
    fun getPriceUpdates(symbol: String): Flow<Price> = flow {
        val webSocket = createWebSocket(symbol)
        try {
            webSocket.messages.collect { message ->
                val price = parsePrice(message)
                emit(price)
            }
        } finally {
            webSocket.close()
        }
    }
    .flowOn(Dispatchers.IO)
    .retry(3) { exception ->
        exception is NetworkException
    }
}

class DashboardService(
    private val userService: UserService,
    private val orderService: OrderService,
    private val notificationService: NotificationService
) {
    fun getDashboardData(userId: String): Flow<DashboardData> = 
        combine(
            userService.getUserFlow(userId),
            orderService.getOrdersFlow(userId),
            notificationService.getNotificationsFlow(userId)
        ) { user, orders, notifications ->
            DashboardData(
                user = user,
                orders = orders,
                notifications = notifications,
                lastUpdated = System.currentTimeMillis()
            )
        }
}

// Традиционный Spring MVC (блокирующий)
@RestController
class UserController(private val userService: UserService) {
    
    @GetMapping("/users/{id}")
    fun getUser(@PathVariable id: String): User {
        return userService.findById(id) // блокирует поток
    }
}

// Spring WebFlux с корутинами (неблокирующий)
@RestController
class UserControllerCoroutines(private val userService: UserService) {
    
    @GetMapping("/users/{id}")
    suspend fun getUser(@PathVariable id: String): User {
        return userService.findById(id) // не блокирует поток
    }
}

@RestController
@RequestMapping("/api/v1")
class ApiController(
    private val userService: UserService,
    private val orderService: OrderService
) {
    
    // Простой suspend endpoint
    @GetMapping("/users/{id}")
    suspend fun getUser(@PathVariable id: String): UserDto {
        return userService.findById(id)
    }
    
    // Параллельная загрузка данных
    @GetMapping("/users/{id}/dashboard")
    suspend fun getUserDashboard(@PathVariable id: String): DashboardDto {
        // Параллельное выполнение запросов
        val userDeferred = async { userService.findById(id) }
        val ordersDeferred = async { orderService.findByUserId(id) }
        val preferencesDeferred = async { preferencesService.findByUserId(id) }
        
        return DashboardDto(
            user = userDeferred.await(),
            orders = ordersDeferred.await(),
            preferences = preferencesDeferred.await()
        )
    }
    
    // Flow как response
    @GetMapping("/users/{id}/notifications", produces = [MediaType.TEXT_EVENT_STREAM_VALUE])
    fun getNotificationStream(@PathVariable id: String): Flow<NotificationDto> {
        return notificationService.getNotificationFlow(id)
            .map { it.toDto() }
    }
}

@ControllerAdvice
class CoroutineExceptionHandler {
    
    @ExceptionHandler(IllegalArgumentException::class)
    suspend fun handleValidationException(ex: IllegalArgumentException): ResponseEntity<ErrorDto> {
        // Может выполнять suspend операции для логирования/метрик
        auditService.logError(ex)
        
        return ResponseEntity.badRequest()
            .body(ErrorDto("Validation failed: ${ex.message}"))
    }
    
    @ExceptionHandler(TimeoutCancellationException::class)
    suspend fun handleTimeout(ex: TimeoutCancellationException): ResponseEntity<ErrorDto> {
        return ResponseEntity.status(HttpStatus.REQUEST_TIMEOUT)
            .body(ErrorDto("Request timeout"))
    }
}

// Репозиторий с suspend функциями
interface UserRepository : CoroutineCrudRepository<User, String> {
    
    // Автоматически генерируемые suspend методы
    // suspend fun findById(id: String): User?
    // suspend fun save(user: User): User
    // suspend fun deleteById(id: String)
    
    // Кастомные suspend запросы
    @Query("SELECT * FROM users WHERE email = :email")
    suspend fun findByEmail(email: String): User?
    
    @Query("SELECT * FROM users WHERE status = :status")
    fun findByStatus(status: UserStatus): Flow<User>
    
    @Query("SELECT * FROM users WHERE created_at > :date")
    suspend fun findRecentUsers(date: LocalDateTime): List<User>
}

// Использование в сервисе
@Service
class UserService(private val userRepository: UserRepository) {
    
    suspend fun createUser(request: CreateUserRequest): User {
        // Проверка существования пользователя
        val existingUser = userRepository.findByEmail(request.email)
        if (existingUser != null) {
            throw UserAlreadyExistsException("User with email ${request.email} already exists")
        }
        
        // Создание нового пользователя
        val user = User(
            id = UUID.randomUUID().toString(),
            email = request.email,
            name = request.name,
            status = UserStatus.ACTIVE,
            createdAt = LocalDateTime.now()
        )
        
        return userRepository.save(user)
    }
    
    // Streaming результатов
    fun getActiveUsersStream(): Flow<User> {
        return userRepository.findByStatus(UserStatus.ACTIVE)
    }
}

// Reactive MongoDB репозиторий
interface ProductRepository : ReactiveMongoRepository<Product, String> {
    
    fun findByCategory(category: String): Flow<Product>
    
    @Query("{ 'price': { \$gte: ?0, \$lte: ?1 } }")
    fun findByPriceRange(minPrice: BigDecimal, maxPrice: BigDecimal): Flow<Product>
}

// Сервис с корутинами
@Service
class ProductService(private val productRepository: ProductRepository) {
    
    suspend fun findProductById(id: String): Product? {
        return productRepository.findById(id).awaitFirstOrNull()
    }
    
    suspend fun saveProduct(product: Product): Product {
        return productRepository.save(product).awaitFirst()
    }
    
    // Batch обработка через Flow
    suspend fun updatePrices(updates: List<PriceUpdate>) {
        updates.asFlow()
            .flatMapConcat { update ->
                productRepository.findById(update.productId)
                    .map { product -> 
                        product.copy(price = update.newPrice)
                    }
                    .flatMap { productRepository.save(it) }
            }
            .collect() // Выполняем все обновления
    }
}

@Configuration
@EnableWebFluxSecurity
class SecurityConfig {
    
    @Bean
    fun securityWebFilterChain(http: ServerHttpSecurity): SecurityWebFilterChain {
        return http
            .authorizeExchange { exchanges ->
                exchanges
                    .pathMatchers("/api/public/**").permitAll()
                    .pathMatchers("/api/admin/**").hasRole("ADMIN")
                    .anyExchange().authenticated()
            }
            .oauth2ResourceServer { oauth2 ->
                oauth2.jwt()
            }
            .build()
    }
}

// Controller с безопасностью
@RestController
@RequestMapping("/api/secure")
class SecureController(private val orderService: OrderService) {
    
    @GetMapping("/orders")
    suspend fun getMyOrders(authentication: Authentication): List<OrderDto> {
        val userId = authentication.name
        return orderService.findByUserId(userId)
    }
    
    @PreAuthorize("hasRole('ADMIN')")
    @GetMapping("/admin/orders")
    suspend fun getAllOrders(): List<OrderDto> {
        return orderService.findAll()
    }
    
    // Получение текущего пользователя в корутине
    @GetMapping("/profile")
    suspend fun getProfile(): UserProfileDto {
        val authentication = currentCoroutineContext()[ReactiveSecurityContextHolder.CONTEXT_KEY]
            ?.authentication ?: throw UnauthorizedException()
        
        return userService.getProfile(authentication.name)
    }
}

@Component
class CoroutineAuthenticationManager(
    private val userService: UserService,
    private val tokenService: TokenService
) : ReactiveAuthenticationManager {
    
    override fun authenticate(authentication: Authentication): Mono<Authentication> {
        return mono {
            authenticateCoroutine(authentication)
        }
    }
    
    private suspend fun authenticateCoroutine(authentication: Authentication): Authentication {
        val token = authentication.credentials as String
        
        // Неблокирующая проверка токена
        val claims = tokenService.validateToken(token)
        val user = userService.findById(claims.userId)
            ?: throw BadCredentialsException("User not found")
        
        return UsernamePasswordAuthenticationToken(
            user.email, 
            null, 
            user.authorities
        )
    }
}

@Configuration
class CoroutineConfig {
    
    @Bean
    @Primary
    fun applicationCoroutineScope(): CoroutineScope {
        return CoroutineScope(
            SupervisorJob() + 
            Dispatchers.Default + 
            CoroutineName("SpringApp") +
            CoroutineExceptionHandler { _, exception ->
                log.error("Unhandled coroutine exception", exception)
            }
        )
    }
    
    @Bean
    fun ioCoroutineScope(): CoroutineScope {
        return CoroutineScope(
            SupervisorJob() + 
            Dispatchers.IO + 
            CoroutineName("SpringIO")
        )
    }
}

// Использование кастомных scope
@Service
class BackgroundTaskService(
    @Qualifier("ioCoroutineScope") private val ioScope: CoroutineScope
) {
    
    fun startBackgroundProcessing() {
        ioScope.launch {
            while (isActive) {
                processDataBatch()
                delay(TimeUnit.MINUTES.toMillis(5))
            }
        }
    }
}

@Component
class CoroutineMetrics(private val meterRegistry: MeterRegistry) {
    
    private val activeCoroutines = Gauge.builder("coroutines.active")
        .description("Number of active coroutines")
        .register(meterRegistry)
    
    private val coroutineTimer = Timer.builder("coroutines.execution.time")
        .description("Coroutine execution time")
        .register(meterRegistry)
    
    suspend fun <T> withMetrics(operation: String, block: suspend () -> T): T {
        return Timer.Sample.start(meterRegistry).use { sample ->
            activeCoroutines.incrementAndGet()
            try {
                block()
            } finally {
                activeCoroutines.decrementAndGet()
                sample.stop(Timer.builder("coroutines.$operation.time").register(meterRegistry))
            }
        }
    }
}

// Использование в контроллере
@RestController
class MetricsController(private val metrics: CoroutineMetrics) {
    
    @GetMapping("/data")
    suspend fun getData(): DataDto = metrics.withMetrics("getData") {
        loadDataFromMultipleSources()
    }
}

@RestController
@RequestMapping("/api/gateway")
class ApiGateway(
    private val userService: UserServiceClient,
    private val orderService: OrderServiceClient,
    private val inventoryService: InventoryServiceClient
) {
    
    @GetMapping("/users/{id}/complete-profile")
    suspend fun getCompleteProfile(@PathVariable id: String): CompleteProfileDto = 
        supervisorScope {
            // Параллельные вызовы микросервисов
            val userProfile = async { 
                try { userService.getProfile(id) }
                catch (e: Exception) { null } // graceful degradation
            }
            
            val orderHistory = async { 
                try { orderService.getOrderHistory(id) }
                catch (e: Exception) { emptyList() }
            }
            
            val recommendations = async { 
                try { inventoryService.getRecommendations(id) }
                catch (e: Exception) { emptyList() }
            }
            
            CompleteProfileDto(
                profile = userProfile.await(),
                orders = orderHistory.await(),
                recommendations = recommendations.await()
            )
        }
}

@Service
class OrderEventProcessor(
    private val orderRepository: OrderRepository,
    private val notificationService: NotificationService,
    private val auditService: AuditService
) : CoroutineScope {
    
    private val job = SupervisorJob()
    override val coroutineContext = Dispatchers.Default + job
    
    private val orderEvents = MutableSharedFlow<OrderEvent>()
    
    @PostConstruct
    fun startProcessing() {
        // Обработка событий заказов
        launch {
            orderEvents.collect { event ->
                when (event) {
                    is OrderEvent.Created -> handleOrderCreated(event)
                    is OrderEvent.Cancelled -> handleOrderCancelled(event)
                    is OrderEvent.Completed -> handleOrderCompleted(event)
                }
            }
        }
    }
    
    suspend fun publishEvent(event: OrderEvent) {
        orderEvents.emit(event)
    }
    
    private suspend fun handleOrderCreated(event: OrderEvent.Created) = supervisorScope {
        // Параллельная обработка
        launch { notificationService.sendOrderConfirmation(event.order) }
        launch { auditService.logOrderCreation(event.order) }
        launch { updateInventory(event.order) }
    }
    
    @PreDestroy
    fun stop() {
        job.cancel()
    }
}

@Service
class DataSyncService(
    private val externalApiClient: ExternalApiClient,
    private val localRepository: DataRepository
) {
    
    @Scheduled(fixedDelay = 300_000) // каждые 5 минут
    suspend fun syncData() {
        log.info("Starting data synchronization")
        
        try {
            externalApiClient.getUpdatedRecords()
                .chunked(100) // обрабатываем пачками по 100
                .collect { batch ->
                    processBatch(batch)
                }
        } catch (e: Exception) {
            log.error("Data sync failed", e)
        }
    }
    
    private suspend fun processBatch(records: List<ExternalRecord>) = supervisorScope {
        records.map { record ->
            async {
                try {
                    val localRecord = record.toLocalRecord()
                    localRepository.save(localRecord)
                } catch (e: Exception) {
                    log.warn("Failed to process record ${record.id}", e)
                }
            }
        }.awaitAll()
    }
}

@Configuration
class WebFluxConfig : WebFluxConfigurer {
    
    override fun configureHttpMessageCodecs(configurer: ServerCodecConfigurer) {
        // Увеличиваем лимиты для streaming
        configurer.defaultCodecs().maxInMemorySize(16 * 1024 * 1024) // 16MB
    }
}

// Настройка R2DBC connection pool
@Configuration
class R2dbcConfig {
    
    @Bean
    fun connectionFactory(): ConnectionFactory {
        return ConnectionFactories.get(
            ConnectionFactoryOptions.builder()
                .option(DRIVER, "postgresql")
                .option(HOST, "localhost")
                .option(PORT, 5432)
                .option(USER, "app")
                .option(PASSWORD, "password")
                .option(DATABASE, "appdb")
                .option(INITIAL_SIZE, 10)
                .option(MAX_SIZE, 50)
                .option(MAX_IDLE_TIME, Duration.ofMinutes(30))
                .build()
        )
    }
}

@Component
@Order(-2) // Высокий приоритет
class GlobalErrorWebExceptionHandler : WebExceptionHandler {
    
    override fun handle(exchange: ServerWebExchange, ex: Throwable): Mono<Void> {
        return when (ex) {
            is CancellationException -> {
                // Корутина была отменена - это нормально
                exchange.response.statusCode = HttpStatus.NO_CONTENT
                exchange.response.setComplete()
            }
            is TimeoutCancellationException -> {
                exchange.response.statusCode = HttpStatus.REQUEST_TIMEOUT
                writeErrorResponse(exchange, "Request timeout")
            }
            else -> {
                exchange.response.statusCode = HttpStatus.INTERNAL_SERVER_ERROR
                writeErrorResponse(exchange, "Internal server error")
            }
        }
    }
    
    private fun writeErrorResponse(exchange: ServerWebExchange, message: String): Mono<Void> {
        val buffer = exchange.response.bufferFactory().wrap(message.toByteArray())
        return exchange.response.writeWith(Mono.just(buffer))
    }
}

// ❌ ПЛОХО - утечка памяти
class UserService {
    fun loadUserData(userId: String) {
        GlobalScope.launch {
            val userData = fetchUserFromApi(userId) // может выполняться вечно
            updateCache(userData)
        }
        // Нет способа отменить эту корутину!
        // Если UserService уничтожается, корутина продолжает работать
    }
}

// ❌ ПЛОХО - долгоживущие ссылки
class MainActivity {
    fun startBackgroundWork() {
        GlobalScope.launch {
            while (true) {
                processData() // бесконечный цикл
                delay(1000)
                // Активность может быть уничтожена, но корутина продолжит работу
                // и будет держать ссылку на контекст через замыкание
            }
        }
    }
}

// ✅ ХОРОШО - контролируемый scope
class UserService : CoroutineScope {
    private val job = SupervisorJob()
    override val coroutineContext = Dispatchers.IO + job
    
    fun loadUserData(userId: String) {
        launch { // привязано к scope сервиса
            val userData = fetchUserFromApi(userId)
            updateCache(userData)
        }
    }
    
    fun cleanup() {
        job.cancel() // отменяет ВСЕ корутины сервиса
    }
}

// ✅ ХОРОШО - scope привязанный к lifecycle
class BackgroundTaskManager {
    private var serviceJob: Job? = null
    
    fun start() {
        serviceJob = CoroutineScope(Dispatchers.Default).launch {
            while (isActive) { // проверяет отмену
                processData()
                delay(1000)
            }
        }
    }
    
    fun stop() {
        serviceJob?.cancel() // явная отмена
        serviceJob = null
    }
}

// ❌ ПЛОХО - утечка ViewModel
class UserViewModel : ViewModel() {
    fun loadData() {
        GlobalScope.launch { // переживёт ViewModel
            val data = repository.loadData()
            // Попытка обновить UI после уничтожения ViewModel
            _uiState.value = UiState.Success(data)
        }
    }
}

// ✅ ХОРОШО - viewModelScope
class UserViewModel : ViewModel() {
    fun loadData() {
        viewModelScope.launch { // автоматически отменяется в onCleared()
            val data = repository.loadData()
            _uiState.value = UiState.Success(data)
        }
    }
}

// ❌ ПЛОХО - блокировка UI
class DataProcessor {
    suspend fun processLargeDataset(data: List<DataItem>) {
        // Выполняется в Main dispatcher по умолчанию
        data.forEach { item ->
            val result = performHeavyComputation(item) // БЛОКИРУЕТ UI!
            updateProgressBar(result.progress)
        }
    }
}

// ❌ ПЛОХО - синхронные операции в Main
suspend fun loadDataFromNetwork(): String {
    // Если это выполняется в Main dispatcher
    return URL("https://api.example.com/data")
        .readText() // Блокирующий I/O вызов! UI зависнет
}

// ✅ ХОРОШО - правильное использование dispatchers
class DataProcessor {
    suspend fun processLargeDataset(data: List<DataItem>) = withContext(Dispatchers.Main) {
        data.forEach { item ->
            // Тяжёлые вычисления в background
            val result = withContext(Dispatchers.Default) {
                performHeavyComputation(item)
            }
            // Обновление UI в Main
            updateProgressBar(result.progress)
        }
    }
}

// ✅ ХОРОШО - правильная архитектура
class NetworkService {
    suspend fun loadData(): String = withContext(Dispatchers.IO) {
        // I/O операции в IO dispatcher
        httpClient.get("https://api.example.com/data")
    }
    
    suspend fun processData(rawData: String): ProcessedData = withContext(Dispatchers.Default) {
        // CPU-интенсивные операции в Default dispatcher
        parseAndProcess(rawData)
    }
}

class UIController {
    suspend fun refreshData() {
        // UI код остаётся в Main dispatcher
        showLoading()
        
        try {
            val rawData = networkService.loadData()
            val processedData = networkService.processData(rawData)
            
            // Обновление UI в Main dispatcher
            displayData(processedData)
        } catch (e: Exception) {
            showError(e.message)
        } finally {
            hideLoading()
        }
    }
}

// ❌ ПЛОХО - блокирующие операции в корутинах
suspend fun badDatabaseAccess() {
    // Блокирующий JDBC в корутине
    val connection = DriverManager.getConnection(jdbcUrl)
    val resultSet = connection.createStatement()
        .executeQuery("SELECT * FROM users") // блокирует поток!
}

// ✅ ХОРОШО - неблокирующие операции
suspend fun goodDatabaseAccess() {
    // R2DBC или другие неблокирующие драйверы
    val users = userRepository.findAll() // suspend функция
    return users
}

// ✅ ХОРОШО - изоляция блокирующих операций
suspend fun legacySystemIntegration() = withContext(Dispatchers.IO) {
    // Изолируем блокирующий код в IO dispatcher
    val legacyClient = LegacyJdbcClient()
    legacyClient.blockingOperation() // ограничено IO pool
}

// launch для side effects
class NotificationService {
    fun sendNotifications(users: List<User>) {
        users.forEach { user ->
            // Не ждём завершения каждой отправки
            scope.launch {
                emailService.sendEmail(user.email, "Welcome!")
                analyticsService.trackEmailSent(user.id)
            }
        }
        // Функция завершается сразу, уведомления отправляются в background
    }
}

// ❌ ПЛОХО - неправильное использование launch
class DataService {
    suspend fun getUserData(id: String): UserData {
        var userData: UserData? = null
        
        launch { // launch НЕ блокирует выполнение
            userData = fetchUserData(id)
        }
        
        return userData!! // ОШИБКА! userData всегда null
    }
}

// async для параллельных операций
class UserProfileService {
    suspend fun loadCompleteProfile(userId: String): CompleteProfile {
        // Запускаем операции параллельно
        val profileDeferred = async { userService.getProfile(userId) }
        val ordersDeferred = async { orderService.getUserOrders(userId) }
        val preferencesDeferred = async { preferencesService.getUserPreferences(userId) }
        
        // Ждём завершения всех операций
        return CompleteProfile(
            profile = profileDeferred.await(),
            orders = ordersDeferred.await(),
            preferences = preferencesDeferred.await()
        )
    }
}

class OrderProcessor {
    // launch для независимых действий
    suspend fun processOrder(order: Order) {
        // Основная логика обработки заказа
        val processedOrder = validateAndProcessOrder(order)
        
        // Независимые side effects - не ждём их завершения
        launch { sendOrderConfirmation(order.customerEmail) }
        launch { updateInventory(order.items) }
        launch { logOrderProcessing(order.id) }
        launch { updateAnalytics(order) }
        
        // Возвращаем результат сразу
        return processedOrder
    }
    
    // async для агрегации данных
    suspend fun createOrderSummary(orderId: String): OrderSummary {
        // Параллельная загрузка связанных данных
        val orderData = async { orderRepository.findById(orderId) }
        val customerData = async { customerRepository.findById(orderData.await().customerId) }
        val paymentData = async { paymentService.getPaymentInfo(orderId) }
        val shippingData = async { shippingService.getShippingInfo(orderId) }
        
        // Ждём все результаты для создания summary
        return OrderSummary(
            order = orderData.await(),
            customer = customerData.await(),
            payment = paymentData.await(),
            shipping = shippingData.await()
        )
    }
}

// RxJava - функциональная цепочка
class RxUserService {
    fun loadUserProfile(userId: String): Single<UserProfile> {
        return userRepository.findById(userId)
            .flatMap { user ->
                ordersRepository.findByUserId(user.id)
                    .map { orders -> UserProfile(user, orders) }
            }
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .timeout(30, TimeUnit.SECONDS)
            .retry(3)
    }
}

// Корутины - императивный подход
class CoroutineUserService {
    suspend fun loadUserProfile(userId: String): UserProfile = withContext(Dispatchers.IO) {
        withTimeout(30_000) {
            retry(3) {
                val user = userRepository.findById(userId)
                val orders = ordersRepository.findByUserId(user.id)
                UserProfile(user, orders)
            }
        }
    }
}

// RxJava - сложно отлаживать
fun complexRxChain(): Observable<Result> {
    return dataSource1.loadData()
        .flatMap { data1 ->
            dataSource2.loadData(data1.id)
                .flatMap { data2 ->
                    dataSource3.processData(data1, data2)
                        .map { processedData ->
                            Result(data1, data2, processedData)
                        }
                }
        }
        .onErrorResumeNext { error ->
            handleError(error)
                .andThen(Observable.just(defaultResult))
        }
}

// Корутины - линейный код
suspend fun complexCoroutineChain(): Result {
    return try {
        val data1 = dataSource1.loadData()
        val data2 = dataSource2.loadData(data1.id)
        val processedData = dataSource3.processData(data1, data2)
        Result(data1, data2, processedData)
    } catch (error: Exception) {
        handleError(error)
        defaultResult
    }
}

// RxJava - сложная обработка ошибок
fun rxErrorHandling(): Single<String> {
    return networkCall()
        .onErrorResumeNext { throwable ->
            when (throwable) {
                is NetworkException -> cacheCall()
                is TimeoutException -> Single.error(UserFriendlyException("Timeout"))
                else -> Single.error(throwable)
            }
        }
        .doOnError { logError(it) }
}

// Корутины - естественная обработка
suspend fun coroutineErrorHandling(): String {
    return try {
        networkCall()
    } catch (e: NetworkException) {
        cacheCall()
    } catch (e: TimeoutException) {
        logError(e)
        throw UserFriendlyException("Timeout")
    } catch (e: Exception) {
        logError(e)
        throw e
    }
}

// RxJava - мощные операторы для потоков
fun rxStreamProcessing(): Observable<ProcessedData> {
    return dataStream
        .debounce(300, TimeUnit.MILLISECONDS)  // защита от частых событий
        .distinctUntilChanged()               // только изменившиеся значения
        .flatMap { data ->
            processData(data).toObservable()
        }
        .scan { previous, current ->          // аккумуляция состояния
            combineData(previous, current)
        }
        .sample(1, TimeUnit.SECONDS)          // сэмплирование каждую секунду
}

// Корутины - требуют больше кода для той же логики
fun coroutineStreamProcessing(): Flow<ProcessedData> {
    return dataFlow
        .debounce(300)
        .distinctUntilChanged()
        .flatMapLatest { data ->
            flow { emit(processData(data)) }
        }
        .scan(initialValue) { previous, current ->
            combineData(previous, current)
        }
        .sample(1000)
}

// RxJava - встроенные стратегии backpressure
fun rxBackpressure(): Flowable<String> {
    return dataProducer
        .onBackpressureBuffer(1000)           // буферизация
        .onBackpressureDrop { dropped ->      // пропуск элементов
            log.warn("Dropped item: $dropped")
        }
        .observeOn(Schedulers.computation(), false, 128) // настройка буфера
}

// ❌ ПЛОХО - неэффективное использование
class BadPerformanceService {
    suspend fun processData(items: List<Item>): List<Result> {
        return items.map { item ->
            // Каждая операция создаёт новый context switch
            withContext(Dispatchers.Default) {
                heavyComputation(item)
            }
        }
    }
}

// ✅ ХОРОШО - батчевая обработка
class GoodPerformanceService {
    suspend fun processData(items: List<Item>): List<Result> = 
        withContext(Dispatchers.Default) {
            // Один context switch для всей обработки
            items.map { item ->
                heavyComputation(item)
            }
        }
}

// ❌ ПЛОХО - создание множества корутин
class IneffientProcessor {
    suspend fun processLargeDataset(data: List<DataItem>) {
        data.forEach { item ->
            launch { // создаём корутину для каждого элемента
                processItem(item)
            }
        }
    }
}

// ✅ ХОРОШО - батчевая обработка
class EfficientProcessor {
    suspend fun processLargeDataset(data: List<DataItem>) {
        data.chunked(100) // группы по 100 элементов
            .map { chunk ->
                async {
                    chunk.forEach { item ->
                        processItem(item) // обрабатываем в одной корутине
                    }
                }
            }
            .awaitAll()
    }
}