Flow — ما هو، التدفقات الباردة والساخنة في كوروتينات Kotlin

ما هو Flow في Kotlin؟

Flow هو نوع من حزمة kotlinx.coroutines.flow، يمثل تدفق بيانات غير متزامن بارد. في جوهره، Flow هو تسلسل كوروتيني يصدر قيمًا عبر دالة emit() وينتهي إما بنجاح أو باستثناء. يتم جمع التدفق عبر العامل الطرفي collect()، وهو دالة suspend.

الدلالات الباردة

التدفق البارد يعني أن الكود داخل مُنشئ flow يُنفذ من جديد لكل مشترك. يتصرف Observable.fromIterable في RxJava بشكل مشابه: المشترك الجديد يتلقى جميع القيم من البداية. في Flow، يتم تنفيذ ذلك عبر دالة suspend collect، التي تحجب الكوروتين طوال مدة جمع البيانات.

منشئات Flow

توفر Kotlin عدة طرق لإنشاء Flow: flow { } — البناء الأساسي مع emit()، flowOf(vararg values) — لمجموعة ثابتة من القيم، .asFlow() — امتداد للمجموعات و Sequence. جميع المنشئات باردة — يتم إنشاء البيانات فقط عند استدعاء العامل الطرفي.

التدفقات الباردة والساخنة

التقسيم إلى تدفقات باردة وساخنة هو مفهوم رئيسي في البرمجة التفاعلية. التدفق البارد (Flow, Observable) يبدأ توليد البيانات عند الاشتراك. التدفق الساخن (Channel, SharedFlow) يصدر البيانات بشكل مستقل — المشترك يتلقى فقط ما يحدث بعد الاشتراك، دون بداية التسلسل.

SharedFlow هو تدفق ساخن يمكن أن يكون له مشتركون متعددون ويعيد تشغيل القيم الأخيرة عند تكوين replay. SharedFlow مناسب للأحداث (إشعارات لمرة واحدة). StateFlow هو نوعه مع قيمة حالة ثابتة، يخزن أحدث قيمة للمشتركين الجدد.

ChannelFlow يستخدم Channel داخليًا، ويجمع خصائص Flow و Channel. يدعم التخزين المؤقت والضغط العكسي عبر capacity. ChannelFlow مفيد عند تحويل واجهات برمجة التطبيقات القائمة على الاستدعاء إلى تدفق تفاعلي، حيث تُصدر القيم من كوروتينات مختلفة.

التحويل بين البارد والساخن

لتحويل التدفق البارد Flow إلى SharedFlow ساخن يُستخدم العامل shareIn(scope, started, replay). تتحكم المعلمة started في لحظة البدء: SharingStarted.WhileSubscribed() — نشط طالما هناك مشتركون، Lazily — البدء عند أول مشترك، Eagerly — البدء الفوري. التحويل العكسي — ساخن إلى بارد: StateFlow.asFlow() يُرجع Flow باردًا يصدر القيمة الحالية لـ StateFlow عند collect. هذا مناسب للاختبار.

عوامل Flow

Flow يوفر مجموعة غنية من العوامل التي تعمل كدوال suspend داخل الكوروتين. العوامل عديمة الحالة وتُرجع Flow جديدًا — التدفق الأصلي يبقى دون تغيير. هذا يسمح ببناء سلاسل تحويل آمنة دون آثار جانبية.

العامل map يحول كل قيمة من التدفق عبر تحويل غير متزامن أو متزامن. filter يمرر فقط القيم التي تحقق الشرط. catch يلتقط الاستثناءات قبل العامل الطرفي ويسمح باستعادة التدفق. flatMapLatest يلغي الإصدار السابق عند وصول قيمة جديدة — مشابه لـ switchMap في Rx.

العامل debounce في Flow يؤخر نشر القيمة لمهلة محددة. إذا وصلت قيمة جديدة خلال هذه المدة، يُعاد ضبط المؤقت. في Android، يُستخدم debounce للبحث: يُرسل الطلب فقط بعد توقف 300-400 مللي ثانية، مما يقلل استدعاءات API بمقدار 3-5 مرات.

العوامل الطرفية

بالإضافة إلى collect()، يدعم Flow عوامل طرفية أخرى: toList() يجمع كل القيم في قائمة — مفيد للاختبارات، first() يُرجع العنصر الأول ويلغي التدفق، single() يتوقع عنصرًا واحدًا بالضبط. fold(initial) يُراكم القيم عبر دالة مُمررة. جميع العوامل الطرفية هي دوال suspend ويجب استدعاؤها داخل كوروتين أو دالة suspend أخرى.

أمثلة كود Flow

المثال الأول — Flow أساسي يولد أرقامًا مع تحويل عبر العامل map:

val numberFlow = flow {
    for (i in 1..5) {
        delay(500)
        emit(i)
    }
}

scope.launch {
    numberFlow
        .map { "الرقم: $it" }
        .collect { value ->
            println(value)
        }
}

المثال الثاني — تحويل التدفق مع التصفية ومعالجة الأخطاء عبر catch:

flow {
    emit("data1")
    emit("data2")
    throw RuntimeException("network error")
}
    .catch { e ->
        emit("fallback_data")
    }
    .collect { value ->
        println(value)
    }

المثال الثالث — استخدام StateFlow في ViewModel لواجهة مستخدم تفاعلية في Jetpack Compose:

class SearchViewModel : ViewModel() {
    private val _query = MutableStateFlow("")
    val results: StateFlow<List<Result>> = _query
        .debounce(300)
        .flatMapLatest { query ->
            repository.search(query)
        }
        .catch { emit(emptyList()) }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), emptyList())

    fun onQueryChanged(query: String) {
        _query.value = query
    }
}

StateFlow و SharedFlow

StateFlow هو تدفق ساخن بقيمة حالية واحدة. يخزن أحدث قيمة ويمررها فورًا لمشترك جديد. StateFlow هو حاوية حالة قابلة للملاحظة، يدعم مقارنة equals — إذا تطابقت القيمة الجديدة مع الحالية، لا يحدث إصدار. يستخدم Jetpack Compose StateFlow عبر collectAsState().

SharedFlow هو تدفق ساخن أكثر مرونة بدون قيمة أولية إلزامية. يُكوّن SharedFlow عبر replay (عدد القيم للمشتركين الجدد)، extraBufferCapacity (المخزن المؤقت خارج إعادة التشغيل)، و onBufferOverflow (استراتيجية عند التدفق الزائد). SharedFlow مثالي للأحداث لمرة واحدة: التنقل، Snackbar، التحليلات.

Flow في بنية Android موصى به من قبل Google كمصدر بيانات رئيسي (الطبقة: Repository → UseCase → ViewModel). LiveData أقل من Flow في المرونة: Flow يدعم الكوروتينات والعوامل والضغط العكسي ويعمل خارج طبقة UI. الترحيل من LiveData إلى Flow هو ممارسة قياسية في مشاريع Android الحديثة.

عند استخدام Flow في ViewModel، من المهم اختيار النوع الصحيح. StateFlow مثالي لحالة UI التي يجب أن تبقى بعد تدوير الشاشة. SharedFlow مناسب للأحداث حيث إعادة المعالجة غير مقبولة — مثل التنقل. Flow مع collect() في lifecycleScope يعطي أقصى تحكم في سياق التنفيذ لكنه يتطلب إلغاءً يدويًا عند مغادرة الشاشة.

يتم اختبار Flow عبر kotlinx-coroutines-test. توفر المكتبة TestDispatcher — وقتًا افتراضيًا يسمح بتسريع التأخيرات (delay) والتحكم في ترتيب تنفيذ الكوروتينات. ينشئ TestScope.runTest { } بيئة معزولة لاختبار Flow. يُستخدم العامل toList() غالبًا في الاختبارات لجمع كل قيم التدفق مع مهلة زمنية، للتحقق من أن التدفق أصدر تسلسل البيانات الصحيح.

يتكامل Flow بشكل جيد مع Room (مكتبة Android لقواعد البيانات): يمكن لدوال DAO إرجاع Flow<List<Entity>>. يصدر Room تلقائيًا قيمة جديدة عند أي تغيير في الجدول — تتحدث واجهة المستخدم دون مشغل يدوي. يتم تنفيذ ذلك عبر InvalidationTracker، الذي يستخدم داخليًا Flow مع callbackFlow. يلغي هذا النهج الحاجة إلى LiveData ويجعل طبقة البيانات موجهة بالكامل نحو الكوروتينات. Jetpack Compose عبر collectAsState() يشترك في StateFlow ويعيد رسم فقط تلك المكونات التي تغيرت بياناتها — وهذا يعطي أداءً لا يمكن تحقيقه مع البنى الموجهة نحو LiveData. DataStore (بديل SharedPreferences) يعيد أيضًا Flow<Preferences>، مما يوفر قراءة تفاعلية لإعدادات التطبيق دون مشغلات تحديث يدوية.

يدعم Flow التواصل بين العمليات عبر kotlinx-coroutines-core على JVM دون مكتبات إضافية. على سبيل المثال، في تطبيقات الخادم على Ktor، يمكن أن يمثل Flow تدفق رسائل WebSocket الواردة. كل رسالة تُصدر في التدفق، تمر عبر التصفية والتجميع عبر العوامل، ويُرسل النتيجة إلى العميل. هذا النهج يحل محل المكتبات التفاعلية مثل Reactor أو RxJava في مشاريع Kotlin.

يتم توفير توافق Flow مع كود RxJava الحالي عبر الوحدة kotlinx-coroutines-rx3. دالة الامتداد Flow.asObservable() تحول Flow إلى Observable من RxJava 3. التحويل العكسي — CompletableSource.asFlow()، Observable.asFlow(). هذا يبسط الترحيل من RxJava إلى الكوروتينات: يمكن إعادة كتابة المشروع على مراحل، مع ترك بعض الطبقات على RxJava. عند التحويل، يجب مراعاة الفرق في الدلالات الباردة/الساخنة: Observable يمكن أن يكون باردًا وساخنًا، Flow دائمًا بارد لـ Flow العادي وساخن لـ SharedFlow.

معالجة الأخطاء واختبار Flow

معالجة الأخطاء في Flow لها خصوصية: إذا حدث استثناء داخل مُنشئ flow قبل العامل الطرفي، يتم تمريره إلى catch. إذا حدث استثناء في عامل بعد المنشئ، يلتقطه catch بعد ذلك العامل. retryWhen يسمح بإعادة محاولة الاشتراك بشرط: إعادة المحاولة عند خطأ الشبكة حتى 3 مرات، ولكن لا يعيد المحاولة عند CancellationException. Flow يزيل الأخطاء المعتمدة على الحالة لأنه لا يخزن حالة — وهذا يبسط التصحيح مقارنة بـ Observable، حيث Subject يخزن حالة داخلية.

اختبار Flow باستخدام kotlinx-coroutines-test يستخدم TestDispatcher لمحاكاة التأخيرات. Turbine هي مكتبة شائعة من المجتمع لاختبار Flow: test { } يشغل Flow، awaitItem() ينتظر القيمة التالية، awaitComplete() ينتظر الاكتمال. تضيف Turbine مهلة زمنية افتراضية، مما يمنع تعليق الاختبارات. لاختبار StateFlow، استخدم .testIn(scope) مع التحقق من القيم بالترتيب الزمني.