Kotlin: شرح الكوروتينات في كوتلن

// جافا: استدعاءات CompletableFuture المتداخلة
CompletableFuture<Order> future = fetchOrder(id)
    .thenCompose(order -> fetchCustomer(order.getCustomerId()))
    .thenApply(customer -> enrichOrder(order, customer))
    .exceptionally(ex -> handleError(ex));

// كوتلن: كود يبدو متسلسلاً، تنفيذ غير متزامن
suspend fun processOrder(id: String): Order {
    val order = fetchOrder(id)           // تعليق، لا حظر
    val customer = fetchCustomer(order.customerId)  // تعليق مرة أخرى
    return enrichOrder(order, customer)
}

// الخيط: خيط واحد لكل مهمة متزامنة (ثقيل)
// 100,000 خيط = OOM (كل خيط ~1 ميجابايت مكدس)

// الكوروتين: خيوط افتراضية خفيفة الوزن
// 100,000 كوروتين = لا مشكلة (كل كوروتين ~بضع مئات من البايت)

fun main() = runBlocking {
    repeat(100_000) {
        launch {  // 100 ألف كوروتين - لا مشكلة!
            delay(1_000)  // تعليق (لا حظر)
            println("Coroutine $it done")
        }
    }
}

// كلمة suspend: هذه الدالة يمكنها تعليق الكوروتين
suspend fun fetchOrder(id: String): Order {
    delay(500)  // محاكاة استدعاء شبكة (غير مانع)
    return Order(id, 299.99, "CONFIRMED")
}

// دوال suspend يمكن استدعاؤها فقط من كوروتينات أو دوال suspend أخرى
suspend fun processOrder(id: String): Order {
    val order = fetchOrder(id)      // نقطة تعليق
    val customer = fetchCustomer(order.customerId)  // نقطة تعليق
    return order.copy(customer = customer)
}

// CoroutineScope: يحدد عمر الكوروتينات
// جميع الكوروتينات المنطلقة في نطاق مرتبطة بعمره

// runBlocking: يحظر الخيط الحالي حتى تكتمل جميع الكوروتينات
runBlocking {
    launch { delay(1_000); println("Done") }
}

// coroutineScope: يعلق (لا يحظر) حتى تكتمل جميع الأبناء
suspend fun fetchAll() = coroutineScope {
    val order = async { fetchOrder("ORD-001") }
    val customer = async { fetchCustomer("CUST-001") }
    Pair(order.await(), customer.await())
}

// نطاق مخصص (مثلاً في صنف)
class OrderService {
    private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())

    fun process(order: Order) {
        scope.launch { /* عمل غير متزامن */ }
    }

    fun shutdown() {
        scope.cancel()  // إلغاء جميع الكوروتينات الأبناء
    }
}

val job = launch {
    println("Working...")
    delay(1_000)
    println("Done")
}

// حالات Job: New -> Active -> Completing -> Completed
//                                    -> Cancelling -> Cancelled
job.cancel()           // طلب إلغاء
job.join()             // انتظار الاكتمال
job.cancelAndJoin()    // إلغاء + انتظار

// Dispatchers.Default: عمل كثيف المعالج (التوازي = أنوية المعالج)
launch(Dispatchers.Default) { computeOrderTax() }

// Dispatchers.IO: عمليات الإدخال/الإخراج المانعة (حتى 64 خيط)
launch(Dispatchers.IO) { fetchDataFromDb() }

// Dispatchers.Main: خيط واجهة المستخدم (Android/Swing)
launch(Dispatchers.Main) { updateUI() }

// مجدول مخصص
val orderDispatcher = Executors.newFixedThreadPool(8).asCoroutineDispatcher()
launch(orderDispatcher) { processOrder() }

// launch: أطلق وانسَ (يرجع Job، لا نتيجة)
val job: Job = launch {
    delay(1_000)
    println("Background work done")
}
job.join()  // انتظار الاكتمال

// async: يرجع Deferred<T> (كائن Job شبيه بالمستقبل)
val deferred: Deferred<Order> = async {
    fetchOrder("ORD-001")
}
val order = deferred.await()  // تعليق حتى تكون النتيجة جاهزة

suspend fun processOrderConcurrently(id: String): EnrichedOrder = coroutineScope {
    // انطلاق بالتوازي ضمن coroutineScope
    val orderDeferred = async { fetchOrder(id) }
    val customerDeferred = async { fetchCustomer(id) }
    val inventoryDeferred = async { checkInventory(id) }

    // انتظار جميع النتائج
    val order = orderDeferred.await()
    val customer = customerDeferred.await()
    val inventory = inventoryDeferred.await()

    EnrichedOrder(order, customer, inventory)
}

// try-catch في الكوروتين
launch {
    try {
        fetchOrder(id)
    } catch (e: Exception) {
        logger.error("Failed to fetch order", e)
    }
}

// CoroutineExceptionHandler
val handler = CoroutineExceptionHandler { _, exception ->
    logger.error("Coroutine error", exception)
}

launch(handler) {
    fetchOrder(id)  // الاستثناءات غير الملتقطة يعالجها المعالج
}

// SupervisorJob: فشل الابن لا يلغي الإخوة
coroutineScope {
    val supervisor = SupervisorJob()
    with(supervisor) {
        launch { throw Exception("Child 1 fails") }  // فقط هذا الابن يفشل
        launch { delay(100); println("Child 2 still runs") }  // الأخ يستمر
    }
}

// ============================================
// OrderProcessor - معالجة غير متزامنة مع الكوروتينات
// الميزة: إثراء الطلبات المتزامن
// ============================================

import kotlinx.coroutines.*

data class Order(val id: String, val total: Double, val customerId: String, var customerName: String? = null)
data class Customer(val id: String, val name: String, val tier: String)

// محاكاة عمليات غير متزامنة
suspend fun fetchOrder(id: String): Order {
    delay(100)  // محاكاة استعلام قاعدة بيانات
    return Order(id, 299.99 + id.substring(4).toInt() * 100, "CUST-${id.substring(4)}")
}

suspend fun fetchCustomer(id: String): Customer {
    delay(150)  // محاكاة استدعاء API
    return Customer(id, "Customer-$id", if (id.endsWith("1")) "VIP" else "STANDARD")
}

suspend fun checkInventory(orderId: String): Boolean {
    delay(80)  // محاكاة فحص المخزون
    return true
}

// معالجة متسلسلة
suspend fun processSequential(id: String): Order {
    val order = fetchOrder(id)          // 100ms
    val customer = fetchCustomer(order.customerId)  // 150ms
    val available = checkInventory(id)  // 80ms
    // الإجمالي: ~330ms
    return order.copy(customerName = customer.name)
}

// معالجة متزامنة
suspend fun processConcurrent(id: String): Order = coroutineScope {
    val orderDeferred = async { fetchOrder(id) }
    val order = orderDeferred.await()

    // جلب العميل والمخزون بالتوازي
    val customerDeferred = async { fetchCustomer(order.customerId) }
    val inventoryDeferred = async { checkInventory(id) }

    val customer = customerDeferred.await()
    val available = inventoryDeferred.await()
    // الإجمالي: ~100ms + ~150ms = ~250ms (العميل والمخزون بالتوازي)

    if (!available) throw RuntimeException("Inventory unavailable for $id")
    order.copy(customerName = "${customer.name} (${customer.tier})")
}

// معالجة دفعية
suspend fun processBatch(orderIds: List<String>): List<Order> = coroutineScope {
    orderIds.map { id ->
        async { processConcurrent(id) }
    }.awaitAll()
}

fun main() = runBlocking {
    val orderIds = listOf("ORD-001", "ORD-002", "ORD-003", "ORD-004", "ORD-005")

    // قياس التسلسلي
    val seqStart = System.currentTimeMillis()
    val seqResults = orderIds.map { processSequential(it) }
    val seqTime = System.currentTimeMillis() - seqStart
    println("Sequential: ${seqTime}ms for ${seqResults.size} orders")

    // قياس المتزامن
    val conStart = System.currentTimeMillis()
    val conResults = processBatch(orderIds)
    val conTime = System.currentTimeMillis() - conStart
    println("Concurrent: ${conTime}ms for ${conResults.size} orders")

    // طباعة النتائج
    println("\n=== Processed Orders ===")
    conResults.forEach { order ->
        println("  ${order.id}: \$${order.total} USD | ${order.customerName}")
    }

    // التزامن المهيكل: خطأ في واحد يلغي الكل
    println("\n=== Error Handling ===")
    try {
        coroutineScope {
            launch { delay(200); println("Task 1 done") }
            launch { delay(100); throw RuntimeException("Task 2 failed!") }
            launch { delay(300); println("Task 3 done") }
        }
    } catch (e: RuntimeException) {
        println("Caught: ${e.message}")
    }
}

Sequential: 1650ms for 5 orders
Concurrent: 370ms for 5 orders

=== Processed Orders ===
  ORD-001: $1100 USD | Customer-CUST-001 (VIP)
  ORD-002: $1200 USD | Customer-CUST-002 (STANDARD)
  ORD-003: $1300 USD | Customer-CUST-003 (STANDARD)
  ORD-004: $1400 USD | Customer-CUST-004 (STANDARD)
  ORD-005: $1500 USD | Customer-CUST-005 (STANDARD)

=== Error Handling ===
Task 1 done
Caught: Task 2 failed!

import kotlinx.coroutines.*

fun main() = runBlocking {
    // launch يبدأ كوروتين ويعود فورًا (Job)
    val job = launch {
        delay(1000L)
        println("Coroutine completed after 1s")
    }

    println("Job is active: ${job.isActive}")
    println("Main continues...")

    // انتظار اكتمال الكوروتين
    job.join()
    println("After join: ${job.isCompleted}")
}

Job is active: true
Main continues...
Coroutine completed after 1s
After join: true

import kotlinx.coroutines.*

fun main() = runBlocking {
    // async يبدأ كوروتين ويعيد Deferred<T>
    val deferred1 = async {
        delay(1000L)
        "Result 1"
    }
    val deferred2 = async {
        delay(500L)
        "Result 2"
    }

    println("Waiting for both...")
    val result1 = deferred1.await()
    val result2 = deferred2.await()

    println("Got: $result1 and $result2")

    // awaitAll لكل الـ Deferreds
    val deferreds = listOf(
        async { delay(100); "A" },
        async { delay(200); "B" },
        async { delay(150); "C" }
    )
    val allResults = deferreds.awaitAll()
    println("All: $allResults")
}

Waiting for both...
Got: Result 1 and Result 2
All: [A, B, C]

import kotlinx.coroutines.*

fun main() = runBlocking {
    // coroutineScope ينتظر جميع الأطفال
    val result = coroutineScope {
        val n1 = async { delay(500); "A" }.await()
        val n2 = async { delay(300); "B" }.await()
        "$n1-$n2"
    }
    println("Structured result: $result")

    // supervisorScope - فشل طفل لا يلغي البقية
    supervisorScope {
        val job1 = launch {
            delay(100)
            println("Child 1 done")
        }
        val job2 = launch {
            delay(50)
            throw RuntimeException("Oops!")
        }
        val job3 = launch {
            delay(200)
            println("Child 3 still runs!")
        }

        try {
            job2.join()
        } catch (e: Exception) {
            println("Caught: ${e.message}")
        }

        job1.join()
        job3.join()
    }
    println("All done")
}

Structured result: A-B
Child 1 done
Caught: Oops!
Child 3 still runs!
All done

import kotlinx.coroutines.*

fun main() = runBlocking {
    val job = launch {
        repeat(10) { i ->
            println("Working $i...")
            delay(500L)
        }
    }

    // إلغاء بعد 1.2 ثانية
    delay(1200L)
    println("Cancelling...")
    job.cancelAndJoin()
    println("Cancelled: ${!job.isActive}")

    // withTimeout - إلغاء تلقائي بعد مهلة
    val result = withTimeoutOrNull(1500L) {
        repeat(5) { i ->
            println("Task $i...")
            delay(400L)
        }
        "Completed"
    }
    println("Result with timeout: $result")
}

Working 0...
Working 1...
Working 2...
Cancelling...
Cancelled: true
Task 0...
Task 1...
Task 2...
Result with timeout: null

import kotlinx.coroutines.*

fun main() = runBlocking {
    // Dispatchers.Default - CPU-bound
    val cpuResult = withContext(Dispatchers.Default) {
        // حساب ثقيل على CPU
        var sum = 0L
        for (i in 1..10_000_000) sum += i
        "Sum: $sum"
    }
    println(cpuResult)

    // Dispatchers.IO - I/O-bound
    val ioResult = withContext(Dispatchers.IO) {
        // محاكاة I/O
        delay(500L)
        "I/O done"
    }
    println(ioResult)

    // Dispatchers.Main - UI (يحتاج تهيئة في Android)
    // مع kotlinx-coroutines-test في وحدة، Main يعمل
    // val mainResult = withContext(Dispatchers.Main) { "UI" }

    // CoroutineScope مخصص
    val customScope = CoroutineScope(Dispatchers.Default + SupervisorJob())
    val deferred = customScope.async {
        delay(200)
        "Custom scope"
    }
    println(deferred.await())
    customScope.cancel()
}

Sum: 50000005000000
I/O done
Custom scope

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    // Flow بارد - تنفيذ عند الجمع
    val numberFlow: Flow<Int> = flow {
        for (i in 1..5) {
            delay(100)
            emit(i)
        }
    }

    // collect - استهلاك كل قيمة
    println("--- First collector ---")
    numberFlow.collect { value ->
        println("Got: $value")
    }

    // map و filter
    val transformedFlow = numberFlow
        .map { it * 10 }
        .filter { it > 20 }

    println("--- Transformed ---")
    transformedFlow.collect { println("Transformed: $it") }

    // reduce - تجميع كل القيم
    val sumFlow = numberFlow.reduce { acc, value -> acc + value }
    println("Sum: $sumFlow")

    // Flow من قائمة
    val listFlow = listOf(1, 2, 3, 4, 5).asFlow()
    listFlow.collect { println("From list: $it") }
}

--- First collector ---
Got: 1
Got: 2
Got: 3
Got: 4
Got: 5
--- Transformed ---
Transformed: 30
Transformed: 40
Transformed: 50
Reduced: 15
From list: 1
From list: 2
From list: 3
From list: 4
From list: 5

import kotlinx.coroutines.*

fun main() = runBlocking {
    // try-catch مع launch
    val job = launch {
        try {
            delay(500)
            throw RuntimeException("Something went wrong")
        } catch (e: Exception) {
            println("Caught: ${e.message}")
        }
    }
    job.join()

    // CoroutineExceptionHandler - معالج عام
    val handler = CoroutineExceptionHandler { _, exception ->
        println("Handler caught: ${exception.message}")
    }

    val scope = CoroutineScope(Dispatchers.Default + SupervisorJob() + handler)
    scope.launch {
        throw IllegalStateException("Handled error")
    }

    delay(100)
    scope.cancel()
    println("Scope cancelled")
}

Caught: Something went wrong
Handler caught: Handled error
Scope cancelled