تدفق Kotlin على نظام Android

في الروتينات المشتركة، التدفّق هو نوع يمكنه إصدار قيم متعددة بالتسلسل، على عكس الدوال المعلّقة التي تعرض قيمة واحدة فقط. على سبيل المثال، يمكنك استخدام تدفق لتلقّي إشعارات مباشرة من قاعدة بيانات.

يتم إنشاء التدفقات استنادًا إلى الروتينات الفرعية ويمكنها توفير قيم متعددة. يُعدّ التدفق من الناحية النظرية سلسلة من البيانات يمكن حسابها بشكل غير متزامن. يجب أن تكون القيم المنبعثة من النوع نفسه. على سبيل المثال، Flow<Int> هو تدفق ينبعث منه قيم أعداد صحيحة.

يشبه التدفق إلى حد كبير Iterator الذي ينتج سلسلة من القيم، ولكنّه يستخدم دوال تعليق لإنتاج القيم واستهلاكها بشكل غير متزامن. وهذا يعني، على سبيل المثال، أنّه يمكن أن يرسل التدفق بأمان طلب شبكة لإنتاج القيمة التالية بدون حظر سلسلة التعليمات الرئيسية.

هناك ثلاثة عناصر متضمّنة في تدفقات البيانات:

• ينتج المُنتِج بيانات تتم إضافتها إلى مصدر البيانات. وبفضل الروتينات المشتركة، يمكن أن تنتج التدفقات أيضًا بيانات بشكل غير متزامن.
• يمكن (اختياري) للوسطاء تعديل كل قيمة يتم إرسالها إلى البث أو البث نفسه.
• يستهلك المستهلك القيم من البث.

في نظام التشغيل Android، يكون المستودع عادةً منتجًا لبيانات واجهة المستخدم، وتكون واجهة المستخدم هي المستهلك الذي يعرض البيانات في النهاية. وفي أحيان أخرى، تكون طبقة واجهة المستخدم منتِجة لأحداث إدخال المستخدم وتستهلكها الطبقات الأخرى من التسلسل الهرمي. تعمل الطبقات بين المنتج والمستهلك عادةً كوسطاء يعدّلون دفق البيانات ليناسب متطلبات الطبقة التالية.

إنشاء سير عمل

لإنشاء مسارات، استخدِم واجهات برمجة التطبيقات الخاصة بأداة إنشاء المسارات. تنشئ دالة الإنشاء flow مسارًا جديدًا يمكنك من خلاله إرسال قيم جديدة يدويًا إلى بث البيانات باستخدام الدالة emit.

في المثال التالي، يسترد مصدر بيانات أحدث الأخبار تلقائيًا على فترات زمنية ثابتة. بما أنّ دالة التعليق لا يمكنها عرض قيم متعدّدة متتالية، ينشئ مصدر البيانات ويعرض تدفقًا لتلبية هذا الشرط. في هذه الحالة، يعمل مصدر البيانات كمنتج.

class NewsRemoteDataSource(
    private val newsApi: NewsApi,
    private val refreshIntervalMs: Long = 5000
) {
    val latestNews: Flow<List<ArticleHeadline>> = flow {
        while (true) {
            val latestNews = newsApi.fetchLatestNews()
            emit(latestNews) // Emits the result of the request to the flow
            delay(refreshIntervalMs) // Suspends the coroutine for some time
        }
    }
}

// Interface that provides a way to make network requests with suspend functions
interface NewsApi {
    suspend fun fetchLatestNews(): List<ArticleHeadline>
}
Flow.kt

يتم تنفيذ أداة إنشاء flow ضمن روتين فرعي. وبالتالي، يستفيد من واجهات برمجة التطبيقات غير المتزامنة نفسها، ولكن تسري بعض القيود:

• تكون المسارات متسلسلة. بما أنّ المنتج في روتين فرعي، عند استدعاء دالة تعليق، يتم تعليق المنتج إلى أن تعرض دالة التعليق قيمة. في المثال، يتم تعليق المنتِج إلى أن يكتمل طلب الشبكة fetchLatestNews. عندها فقط يتم إرسال النتيجة إلى البث.
• باستخدام أداة إنشاء flow، لا يمكن للمنتج emit قيمًا من CoroutineContext مختلف. لذلك، لا تستدعِ emit في CoroutineContext مختلف من خلال إنشاء إجراءات فرعية جديدة أو باستخدام كتل الرموز withContext. يمكنك استخدام أدوات إنشاء سير عمل أخرى، مثل callbackFlow، في هذه الحالات.

تعديل البث

يمكن للوسطاء استخدام عوامل التشغيل الوسيطة لتعديل تدفق البيانات بدون استهلاك القيم. هذه العوامل هي دوال، وعند تطبيقها على مجموعة من البيانات، يتم إعداد سلسلة من العمليات التي لا يتم تنفيذها إلا عند استخدام القيم في المستقبل. يمكنك الاطّلاع على مزيد من المعلومات حول عوامل التشغيل الوسيطة في مستندات مرجع Flow.

في المثال أدناه، تستخدم طبقة المستودع عامل التشغيل الوسيط map لتحويل البيانات التي سيتم عرضها على View:

class NewsRepository(
    private val newsRemoteDataSource: NewsRemoteDataSource,
    private val userData: UserData
) {
    /**
     * Returns the favorite latest news applying transformations on the flow.
     * These operations are lazy and don't trigger the flow. They just transform
     * the current value emitted by the flow at that point in time.
     */
    val favoriteLatestNews: Flow<List<ArticleHeadline>> =
        newsRemoteDataSource.latestNews
            // Intermediate operation to filter the list of favorite topics
            .map { news -> news.filter { userData.isFavoriteTopic(it) } }
            // Intermediate operation to save the latest news in the cache
            .onEach { news -> saveInCache(news) }
}
Flow.kt

يمكن تطبيق عوامل التشغيل الوسيطة الواحدة تلو الأخرى، ما يؤدي إلى إنشاء سلسلة من العمليات التي يتم تنفيذها بشكل غير مباشر عند إصدار عنصر في التدفق. يُرجى العِلم أنّ مجرد تطبيق عامل وسيط على بث لا يؤدي إلى بدء عملية جمع البيانات من البث.

الجمع من مسار

استخدِم عامل تشغيل نهائي لتفعيل المسار وبدء الاستماع إلى القيم. للحصول على جميع القيم في البث عند إرسالها، استخدِم collect. يمكنك الاطّلاع على مزيد من المعلومات حول عوامل تشغيل المحطة الطرفية في مستندات Flow الرسمية.

بما أنّ collect هي دالة تعليق، يجب تنفيذها ضمن روتين فرعي. تتلقّى هذه الدالة دالة lambda كمعلَمة يتم استدعاؤها لكل قيمة جديدة. بما أنّها دالة تعليق، يمكن أن يتم تعليق الروتين الفرعي الذي يستدعي collect إلى أن يتم إغلاق تدفق البيانات.

بالاستناد إلى المثال السابق، إليك عملية تنفيذ بسيطة للرمز ViewModel الذي يستهلك البيانات من طبقة المستودع:

class LatestNewsViewModel(
    private val newsRepository: NewsRepository
) : ViewModel() {

    init {
        viewModelScope.launch {
            // Trigger the flow and consume its elements using collect
            newsRepository.favoriteLatestNews.collect { favoriteNews ->
                // Update UI with the latest favorite news
            }
        }
    }
}
Flow.kt

يؤدي جمع التدفق إلى تشغيل المنتج الذي يعيد تحميل آخر الأخبار ويعرض نتيجة طلب الشبكة على فترات زمنية ثابتة. بما أنّ المنتج يظل نشطًا دائمًا مع حلقة while(true)، سيتم إغلاق بث البيانات عندما تتم إزالة ViewModel وإلغاء viewModelScope.

يمكن أن يتوقف جمع البيانات في مسار الإحالة للأسباب التالية:

• يتم إلغاء الكوروتين الذي يجمع البيانات، كما هو موضّح في المثال السابق. يؤدي هذا الإجراء أيضًا إلى إيقاف المنتج الأساسي.
• ينهي المنتج إرسال العناصر. في هذه الحالة، يتم إغلاق دفق البيانات ويستأنف الروتين الفرعي الذي استدعى collect التنفيذ.

تكون التدفقات باردة وكسولة ما لم يتم تحديدها باستخدام عوامل تشغيل وسيطة أخرى. وهذا يعني أنّه يتم تنفيذ رمز المنتج في كل مرة يتم فيها استدعاء عامل تشغيل نهائي في التدفق. في المثال السابق، يؤدي توفّر عدة أدوات لجمع البيانات إلى أن يجلب مصدر البيانات آخر الأخبار عدة مرات على فترات زمنية ثابتة مختلفة. لتحسين ومشاركة مسار عندما يجمع عدة مستهلكين البيانات في الوقت نفسه، استخدِم عامل التشغيل shareIn.

رصد الاستثناءات غير المتوقّعة

يمكن أن يتم تنفيذ المنتج من خلال مكتبة تابعة لجهة خارجية. وهذا يعني أنّه يمكن أن يعرض استثناءات غير متوقّعة. للتعامل مع هذه الاستثناءات، استخدِم عامل التشغيل الوسيط catch.

class LatestNewsViewModel(
    private val newsRepository: NewsRepository
) : ViewModel() {

    init {
        viewModelScope.launch {
            newsRepository.favoriteLatestNews
                // Intermediate catch operator. If an exception is thrown,
                // catch and update the UI
                .catch { exception -> notifyError(exception) }
                .collect { favoriteNews ->
                    // Update UI with the latest favorite news
                }
        }
    }
}
Flow.kt

في المثال السابق، عند حدوث استثناء، لا يتم استدعاء دالة collect lambda، لأنّه لم يتم تلقّي عنصر جديد.

يمكن لـ catch أيضًا emit عناصر إلى التدفق. يمكن أن emit طبقة مستودع المثال القيم المخزّنة مؤقتًا بدلاً من ذلك:

class NewsRepository(
    // ...
) {
    val favoriteLatestNews: Flow<List<ArticleHeadline>> =
        newsRemoteDataSource.latestNews
            .map { news -> news.filter { userData.isFavoriteTopic(it) } }
            .onEach { news -> saveInCache(news) }
            // If an error happens, emit the last cached values
            .catch { exception -> emit(lastCachedNews()) }
}
Flow.kt

في هذا المثال، عند حدوث استثناء، يتم استدعاء دالة collect Lambda، وذلك لأنّه تم إرسال عنصر جديد إلى الدفق بسبب الاستثناء.

التنفيذ في CoroutineContext مختلف

بشكل تلقائي، يتم تنفيذ منتج أداة إنشاء flow في CoroutineContext الخاص بالكوروتين الذي يجمع البيانات منه، وكما ذكرنا سابقًا، لا يمكنه emit القيم من CoroutineContext مختلف. قد يكون هذا السلوك غير مرغوب فيه في بعض الحالات. على سبيل المثال، في الأمثلة المستخدَمة في هذا الموضوع، يجب ألا تنفّذ طبقة المستودع عمليات على Dispatchers.Main التي تستخدمها viewModelScope.

لتغيير CoroutineContext لأحد المسارات، استخدِم عامل التشغيل الوسيط flowOn. تغيّر flowOn CoroutineContext للتسلسل المتّجه للأعلى، ما يعني أنّ المنتج وأي عوامل تشغيل وسيطة يتم تطبيقها قبل (أو أعلى) flowOn. لا يتأثر مسار البيانات النهائية (عوامل التشغيل الوسيطة بعد flowOn بالإضافة إلى المستهلك) ويتم تنفيذه على CoroutineContext المستخدَم collect من المسار. في حال توفّر عدة عوامل تشغيل flowOn، يغيّر كل عامل منها المصدر من موقعه الحالي.

class NewsRepository(
    private val newsRemoteDataSource: NewsRemoteDataSource,
    private val userData: UserData,
    private val defaultDispatcher: CoroutineDispatcher
) {
    val favoriteLatestNews: Flow<List<ArticleHeadline>> =
        newsRemoteDataSource.latestNews
            .map { news -> // Executes on the default dispatcher
                news.filter { userData.isFavoriteTopic(it) }
            }
            .onEach { news -> // Executes on the default dispatcher
                saveInCache(news)
            }
            // flowOn affects the upstream flow ↑
            .flowOn(defaultDispatcher)
            // the downstream flow ↓ is not affected
            .catch { exception -> // Executes in the consumer's context
                emit(lastCachedNews())
            }
}
Flow.kt

باستخدام هذا الرمز، يستخدِم عاملَا التشغيل onEach وmap الرمز defaultDispatcher، بينما يتم تنفيذ عامل التشغيل catch والمستهلك على Dispatchers.Main الذي يستخدمه viewModelScope.

بما أنّ طبقة مصدر البيانات تنفّذ عمليات إدخال وإخراج، عليك استخدام أداة إرسال محسَّنة لعمليات الإدخال والإخراج:

class NewsRemoteDataSource(
    // ...
    private val ioDispatcher: CoroutineDispatcher
) {

    val latestNews: Flow<List<ArticleHeadline>> = flow {
        // Executes on the IO dispatcher
        // ...
    }
        .flowOn(ioDispatcher)
}
Flow.kt

تدفّقات البيانات في مكتبات Jetpack

تم دمج Flow في العديد من مكتبات Jetpack، وهو شائع بين مكتبات Android التابعة لجهات خارجية. تُعدّ Flow مناسبة تمامًا لتعديلات البيانات المباشرة والتدفقات اللانهائية من البيانات.

يمكنك استخدام Flow with Room لتلقّي إشعارات بشأن التغييرات في قاعدة البيانات. عند استخدام كائنات الوصول إلى البيانات (DAO)، عليك عرض نوع Flow للحصول على التحديثات الحية.

@Dao
abstract class ExampleDao {
    @Query("SELECT * FROM Example")
    abstract fun getExamples(): Flow<List<Example>>
}
Flow.kt

في كل مرة يحدث فيها تغيير في جدول Example، يتم إصدار قائمة جديدة تتضمّن العناصر الجديدة في قاعدة البيانات.

تحويل واجهات برمجة التطبيقات المستندة إلى عمليات رد الاتصال إلى مسارات

callbackFlow هي أداة إنشاء سير عمل تتيح لك تحويل واجهات برمجة التطبيقات المستندة إلى عمليات رد الاتصال إلى سير عمل. على سبيل المثال، تستخدم واجهات برمجة التطبيقات لنظام التشغيل Android في Firebase Firestore عمليات رد الاتصال.

لتحويل واجهات برمجة التطبيقات هذه إلى تدفّقات والاستماع إلى تعديلات قاعدة بيانات Firestore، يمكنك استخدام الرمز التالي:

class FirestoreUserEventsDataSource(
    private val firestore: FirebaseFirestore
) {
    // Method to get user events from the Firestore database
    fun getUserEvents(): Flow<UserEvents> = callbackFlow {

        // Reference to use in Firestore
        var eventsCollection: CollectionReference? = null
        try {
            eventsCollection = FirebaseFirestore.getInstance()
                .collection("collection")
                .document("app")
        } catch (e: Throwable) {
            // If Firebase cannot be initialized, close the stream of data
            // flow consumers will stop collecting and the coroutine will resume
            close(e)
        }

        // Registers callback to firestore, which will be called on new events
        val subscription = eventsCollection?.addSnapshotListener { snapshot, _ ->
            if (snapshot == null) { return@addSnapshotListener }
            // Sends events to the flow! Consumers will get the new events
            try {
                trySend(snapshot.getEvents())
            } catch (e: Throwable) {
                // Event couldn't be sent to the flow
            }
        }

        // The callback inside awaitClose will be executed when the flow is
        // either closed or cancelled.
        // In this case, remove the callback from Firestore
        awaitClose { subscription?.remove() }
    }
}

على عكس أداة إنشاء flow، تتيح callbackFlow إرسال القيم من CoroutineContext مختلف باستخدام الدالة send أو خارج روتين فرعي باستخدام الدالة trySend.

داخليًا، تستخدم callbackFlow قناة، وهي تشبه من الناحية النظرية قائمة الانتظار. يتم ضبط قناة باستخدام سعة، وهي الحد الأقصى لعدد العناصر التي يمكن تخزينها مؤقتًا. تتضمّن القناة التي تم إنشاؤها في callbackFlow سعة تلقائية تبلغ 64 عنصرًا. عند محاولة إضافة عنصر جديد إلى قناة ممتلئة، يعلّق send عملية الإنتاج إلى أن تتوفّر مساحة للعنصر الجديد، بينما لا يضيف trySend العنصر إلى القناة ويعرض false على الفور.

تضيف الدالة trySend العنصر المحدّد إلى القناة على الفور، وذلك فقط إذا لم يخالف هذا الإجراء قيود السعة، ثم تعرض النتيجة الناجحة.

مراجع إضافية حول مسارات المستخدمين

• اختبار مسارات Kotlin على Android
• ‫StateFlow وSharedFlow
• مراجع إضافية حول الكوروتينات والتدفق في Kotlin