בתהליך של Kotlin ב-Android

בקורוטינות, Flow הוא סוג שיכול להפיק כמה ערכים ברצף, בניגוד לפונקציות השהיה שמחזירות רק ערך אחד. לדוגמה, אתם יכולים להשתמש בתהליך כדי לקבל עדכונים בזמן אמת ממסד נתונים.

הפונקציות האלה מבוססות על קורוטינות ויכולות לספק כמה ערכים. מבחינה קונספטואלית, זרימה היא זרם של נתונים שאפשר לבצע עליהם חישובים באופן אסינכרוני. הערכים שמוחזרים צריכים להיות מאותו סוג. לדוגמה, Flow<Int> הוא תהליך שפולט ערכים מספריים שלמים.

ה-Flow דומה מאוד ל-Iterator שמייצר רצף של ערכים, אבל הוא משתמש בפונקציות השהיה כדי לייצר ולצרוך ערכים באופן אסינכרוני. לדוגמה, התהליך יכול לבצע בבטחה בקשה לאחזור מהרשת כדי ליצור את הערך הבא בלי לחסום את השרשור הראשי.

יש שלושה ישויות שמעורבות במקורות נתונים:

• יצרן יוצר נתונים שנוספים לזרם. בזכות קורוטינות, אפשר גם ליצור נתונים באופן אסינכרוני באמצעות רצפי פעולות.
• (אופציונלי) מתווכים יכולים לשנות כל ערך שמועבר לזרם או את הזרם עצמו.
• צרכן צורך את הערכים מהזרם.

ב-Android, מאגר בדרך כלל מעביר בסטרימינג נתוני ממשק משתמש ממסדי נתונים מקומיים (Room) או משרתי עורף בענן (כמו Firebase), וממשק המשתמש פועל כצרכן של הנתונים. לעומת זאת, ממשק המשתמש יכול ליצור אירועי קלט של משתמשים כדי ששכבות אחרות יוכלו להשתמש בהם. שכבות ביניים (כמו מופעים של ViewModel) משנות את הסטרים כדי לעמוד בדרישות של כל שכבה.

יצירת רצף פעולות

כדי ליצור רצפי פעולות, משתמשים בממשקי ה-API של כלי בניית רצפי פעולות. פונקציית ה-builder‏ 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 מופעל בתוך קורוטינה. לכן, הוא נהנה מאותם ממשקי API אסינכרוניים, אבל יש כמה מגבלות:

• רצפי פעולות הם עוקבים. מכיוון שהפונקציה producer נמצאת בקורוטינה, כשמפעילים פונקציית השהיה, הפונקציה producer מושהית עד שהפונקציה suspend מחזירה ערך. בדוגמה, הבעלים של השירות המנוהל מושעה עד להשלמת בקשה לאחזור מהרשת 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. במסמכי התיעוד הרשמיים של תהליך העבודה מפורט מידע נוסף על אופרטורים של מסוף.

מכיוון ש-collect היא פונקציית השהיה, צריך להריץ אותה בתוך קורוטינה. היא מקבלת פונקציית lambda כפרמטר, שמופעלת על כל ערך חדש. מכיוון שזו פונקציית השהיה, יכול להיות שהקורוטינה שקוראת ל-collect תושהה עד לסגירת ה-Flow.

בהמשך לדוגמה הקודמת, הנה הטמעה פשוטה של 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

האיסוף של התהליך מפעיל את היצרן שמרענן את החדשות האחרונות ומפיק את התוצאה של בקשה לאחזור מהרשת במרווח קבוע. ה-Producer נשאר תמיד פעיל בלולאה while(true), ולכן זרם הנתונים ייסגר כש-ViewModel ינוקה ו-viewModelScope יבוטל.

יכול להיות שהאיסוף של פידים יופסק מהסיבות הבאות:

• שגרת המשך (coroutine) שאוספת מבוטלת, כמו בדוגמה הקודמת. הפעולה הזו גם מפסיקה את ההפקה הבסיסית.
• הבעלים של השירות המנוהל מסיים לשדר פריטים. במקרה כזה, זרם הנתונים נסגר והקורוטינה שקראה ל-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

בדוגמה הקודמת, כשמתרחש חריג, פונקציית ה-lambda‏ collect לא נקראת, כי לא התקבל פריט חדש.

אפשר גם catchemit פריטים לזרימה. שכבת המאגר לדוגמה יכולה 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

בדוגמה הזו, כשמתרחשת חריגה, פונקציית ה-lambda‏ collect נקראת, כי פריט חדש נפלט לזרם בגלל החריגה.

הרצה בהקשר שונה של קורוטינה

כברירת מחדל, הפונקציה ליצירת flow מופעלת ב-CoroutineContext של שגרת המשך (coroutine) שאוספת ממנה, וכמו שצוין קודם, היא לא יכולה 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.

מכיוון ששכבת מקור הנתונים מבצעת פעולות קלט/פלט, צריך להשתמש ב-dispatcher שעבר אופטימיזציה לפעולות קלט/פלט:

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

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

תזרימים בספריות ובמקורות נתונים של Jetpack

‫Flow משולב בהרבה ספריות של Jetpack, והוא פופולרי בספריות של צד שלישי ובערכות SDK בענן של Android. התכונה 'זרימה' מתאימה מאוד לעדכונים של נתונים בזמן אמת ולזרמים אינסופיים של נתונים.

אתם יכולים להשתמש ב-Flow עם Room כדי לקבל הודעה על שינויים במסד נתונים. כשמשתמשים באובייקטים של גישה לנתונים (DAO), מחזירים סוג Flow כדי לקבל עדכונים בזמן אמת.

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

בכל פעם שיש שינוי בטבלה 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 (נתוני ענן מרחוק) כדי לספק חוויית משתמש בזמן אמת שמתמקדת בשימוש אופליין.

המרת ממשקי API מבוססי-callback ל-Flows

‫callbackFlow הוא כלי ליצירת תהליכים שמאפשר להמיר ממשקי API מבוססי-callback לתהליכים. זה שימושי במיוחד כשעובדים עם מאזינים לכמה אירועים בערכות SDK כמו Cloud Firestore או הגדרת תצורה מרחוק ב-Firebase.

כדי להמיר API של מאזין מבוסס-callback ל-Flow – למשל, כדי להאזין לעדכונים במסד נתונים של 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() }
    }
}

בניגוד לפונקציית ה-builder‏ flow, הפונקציה callbackFlow מאפשרת להפיק ערכים מ-CoroutineContext אחר באמצעות הפונקציה send או מחוץ לקורוטינה באמצעות הפונקציה trySend.

באופן פנימי, callbackFlow משתמש בערוץ, שדומה מאוד מבחינה קונספטואלית לתור חוסם. ערוץ מוגדר עם קיבולת, מספר האלמנטים המקסימלי שאפשר לשמור במאגר. הקיבולת של הערוץ שנוצר ב-callbackFlow היא 64 אלמנטים כברירת מחדל. כשמנסים להוסיף רכיב חדש לערוץ מלא, send משעה את ההפקה עד שיהיה מקום לרכיב החדש, ואילו trySend לא מוסיף את הרכיב לערוץ ומחזיר את false באופן מיידי.

‫trySend מוסיף מיד את האלמנט שצוין לערוץ, רק אם הפעולה הזו לא מפרה את מגבלות הקיבולת שלו, ואז מחזיר את התוצאה של הפעולה המוצלחת.

מקורות מידע נוספים על תהליכי עבודה

• בדיקת תהליכים ב-Kotlin ב-Android
• ‫StateFlow ו-SharedFlow
• Flows