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

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

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

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

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

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

الكيانات المشاركة في تدفّقات البيانات: المستهلك والوسطاء الاختياريون والمنتج
الشكل 1. الجهات المعنية بتدفقات البيانات: المستهلك والوسطاء الاختياريون والمنتج

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

إنشاء سير عمل

لإنشاء مسارات، استخدِم واجهات برمجة التطبيقات الخاصة بأداة إنشاء المسارات. تنشئ دالة الإنشاء 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 ضمن روتين فرعي. وبالتالي، تستفيد من واجهات برمجة التطبيقات غير المتزامنة نفسها، ولكن تنطبق بعض القيود:

  • المسارات متسلسلة. بما أنّ المنتج في روتين فرعي، فعند استدعاء دالة تعليق، يتم تعليق المنتج إلى أن تعرض دالة التعليق قيمة. في المثال، يتم تعليق المنتِج إلى أن يكتمل طلب الشبكة 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) }
}

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

الجمع من مسار

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

بما أنّ 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
            }
        }
    }
}

يؤدي جمع التدفق إلى تشغيل المنتِج الذي يعيد تحميل آخر الأخبار ويعرض نتيجة طلب الشبكة على فترات زمنية ثابتة. بما أنّ المنتج يظل نشطًا دائمًا مع حلقة 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
                }
        }
    }
}

في المثال السابق، عند حدوث استثناء، لا يتم استدعاء دالة 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()) }
}

في هذا المثال، عند حدوث استثناء، يتم استدعاء دالة 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())
            }
}

باستخدام هذا الرمز، يستخدِم عاملَا التشغيل 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)
}

مصادر البيانات وعمليات نقل البيانات في مكتبات Jetpack

تم دمج Flow في العديد من مكتبات Jetpack، وهو شائع في مكتبات Android التابعة لجهات خارجية وحِزم تطوير البرامج (SDK) السحابية. تتلاءم Flow بشكل كبير مع تعديلات البيانات المباشرة وتدفّق البيانات بلا توقّف.

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

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

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

وبالمثل، توفّر مصادر بيانات السحابة الإلكترونية في الوقت الفعلي، مثل Cloud Firestore، إضافات Flow تبث آخر الأخبار مباشرةً إلى طبقة المستودع:

class RemoteMessagesDataSource(
    private val firestore: FirebaseFirestore
) {
    // Returns a Flow emitting updated message lists when collection changes
    val messages: Flow<List<Message>> =
        firestore.collection("messages").dataObjects()
}

يتوافق ذلك بشكل طبيعي مع التخزين على الجهاز: يمكن أن يجمع المستودع التدفقات من كل من Room (ذاكرة التخزين المؤقت المحلية) وFirebase (بيانات السحابة الإلكترونية البعيدة) لتوفير تجربة مستخدم في الوقت الفعلي تعمل في وضع عدم الاتصال بالإنترنت.

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

callbackFlow هي أداة إنشاء سير عمل تتيح لك تحويل واجهات برمجة التطبيقات المستندة إلى عمليات رد الاتصال إلى عمليات سير عمل. ويكون ذلك مفيدًا بشكل خاص عند استخدام متتبِّعات أحداث متعددة في حِزم تطوير البرامج (SDK)، مثل Cloud Firestore أو الإعداد عن بُعد عبر Firebase.

لتحويل واجهة برمجة تطبيقات مستمع مستندة إلى معاودة الاتصال إلى تدفق، مثل الاستماع إلى تحديثات قاعدة بيانات 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 العنصر المحدّد إلى القناة على الفور، ولكن فقط إذا لم يخالف ذلك قيود السعة، ثم تعرض النتيجة الناجحة.

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