Kotlin در اندروید جریان دارد، Kotlin در اندروید جریان دارد

در کوروتین‌ها، یک جریان نوعی است که می‌تواند چندین مقدار را به صورت متوالی منتشر کند، برخلاف توابع suspend که فقط یک مقدار واحد را برمی‌گردانند. به عنوان مثال، می‌توانید از یک جریان برای دریافت به‌روزرسانی‌های زنده از یک پایگاه داده استفاده کنید.

جریان‌ها (Flows) بر روی کوروتین‌ها (Coroutines) ساخته می‌شوند و می‌توانند چندین مقدار ارائه دهند. یک جریان (Flow) از نظر مفهومی، جریانی از داده‌ها است که می‌تواند به صورت ناهمگام (asynchronously) محاسبه شود. مقادیر منتشر شده باید از یک نوع باشند. به عنوان مثال، یک Flow<Int> جریانی است که مقادیر صحیح (integer) منتشر می‌کند.

یک جریان (flow) بسیار شبیه به یک Iterator است که دنباله‌ای از مقادیر را تولید می‌کند، اما از توابع suspend برای تولید و مصرف مقادیر به صورت غیرهمزمان استفاده می‌کند. این به این معنی است که، برای مثال، جریان می‌تواند با خیال راحت یک درخواست شبکه برای تولید مقدار بعدی بدون مسدود کردن نخ اصلی ارسال کند.

سه موجودیت در جریان داده‌ها دخیل هستند:

• یک تولیدکننده (producer) داده‌هایی را تولید می‌کند که به جریان (stream) اضافه می‌شوند. به لطف کوروتین‌ها (Coroutines)، جریان‌ها (flows) می‌توانند داده‌ها را به صورت ناهمگام (asynchronously) نیز تولید کنند.
• (اختیاری) واسطه‌ها می‌توانند هر مقدار منتشر شده در جریان یا خود جریان را تغییر دهند.
• یک مصرف‌کننده، مقادیر را از جریان مصرف می‌کند.

در اندروید، یک مخزن معمولاً تولیدکننده‌ی داده‌های رابط کاربری است که رابط کاربری (UI) را به عنوان مصرف‌کننده‌ای دارد که در نهایت داده‌ها را نمایش می‌دهد. در موارد دیگر، لایه‌ی رابط کاربری تولیدکننده‌ی رویدادهای ورودی کاربر است و سایر لایه‌های سلسله مراتب آنها را مصرف می‌کنند. لایه‌های بین تولیدکننده و مصرف‌کننده معمولاً به عنوان واسطه‌هایی عمل می‌کنند که جریان داده‌ها را تغییر می‌دهند تا آن را با الزامات لایه‌ی بعدی تنظیم کنند.

ایجاد جریان

برای ایجاد جریان‌ها، از APIهای سازنده جریان استفاده کنید. تابع سازنده flow ، یک جریان جدید ایجاد می‌کند که در آن می‌توانید با استفاده از تابع emit مقادیر جدید را به صورت دستی به جریان داده‌ها ارسال کنید.

در مثال زیر، یک منبع داده آخرین اخبار را به طور خودکار در یک بازه زمانی ثابت دریافت می‌کند. از آنجایی که یک تابع suspend نمی‌تواند چندین مقدار متوالی را برگرداند، منبع داده یک جریان ایجاد کرده و برمی‌گرداند تا این الزام را برآورده کند. در این حالت، منبع داده به عنوان تولیدکننده عمل می‌کند.

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های ناهمزمان بهره می‌برد، اما برخی محدودیت‌ها اعمال می‌شود:

• جریان‌ها ترتیبی هستند. از آنجایی که تولیدکننده در یک کوروتین قرار دارد، هنگام فراخوانی یک تابع suspend، تولیدکننده تا زمان بازگشت تابع suspend به حالت تعلیق در می‌آید. در این مثال، تولیدکننده تا زمان تکمیل درخواست شبکه fetchLatestNews به حالت تعلیق در می‌آید. تنها در آن زمان نتیجه به جریان ارسال می‌شود.
• با استفاده از flow builder، تولیدکننده نمی‌تواند مقادیر را از یک CoroutineContext متفاوت emit . بنابراین، با ایجاد coroutine های جدید یا با استفاده از بلوک‌های کد withContext ، emit در یک CoroutineContext متفاوت فراخوانی نکنید. در این موارد می‌توانید از flow builder های دیگری مانند 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 یک تابع suspend است، باید درون یک کوروتین اجرا شود. این تابع یک لامبدا را به عنوان پارامتری می‌گیرد که با هر مقدار جدید فراخوانی می‌شود. از آنجایی که این یک تابع suspend است، کوروتینی که 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 لامبدا فراخوانی نمی‌شود، زیرا هنوز آیتم جدیدی دریافت نشده است.

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 فراخوانی می‌شود، زیرا به دلیل استثنا، یک آیتم جدید به جریان (stream) ارسال شده است.

اجرا در یک CoroutineContext متفاوت

به طور پیش‌فرض، تولیدکننده‌ی یک سازنده‌ی flow ، در CoroutineContext کوروتینی که از آن جمع‌آوری می‌کند، اجرا می‌شود و همانطور که قبلاً ذکر شد، نمی‌تواند مقادیری را از یک CoroutineContext متفاوت emit . این رفتار ممکن است در برخی موارد نامطلوب باشد. به عنوان مثال، در مثال‌های استفاده شده در سراسر این مبحث، لایه‌ی مخزن نباید عملیاتی را روی 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 و consumer روی 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

جریان‌ها در کتابخانه‌های جت‌پک

Flow با بسیاری از کتابخانه‌های Jetpack یکپارچه شده است و در بین کتابخانه‌های شخص ثالث اندروید محبوب است. Flow برای به‌روزرسانی‌های زنده داده‌ها و جریان‌های بی‌پایان داده‌ها بسیار مناسب است.

شما می‌توانید از Flow به همراه Room برای مطلع شدن از تغییرات در پایگاه داده استفاده کنید. هنگام استفاده از اشیاء دسترسی به داده (DAO) ، برای دریافت به‌روزرسانی‌های زنده، نوع Flow برگردانید.

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

هر بار که تغییری در جدول Example ایجاد شود، یک لیست جدید با موارد جدید در پایگاه داده منتشر می‌شود.

تبدیل APIهای مبتنی بر فراخوانی به جریان‌ها

callbackFlow یک سازنده جریان است که به شما امکان می‌دهد APIهای مبتنی بر callback را به جریان تبدیل کنید. به عنوان مثال، APIهای اندروید Firebase Firestore از callbackها استفاده می‌کنند.

برای تبدیل این APIها به جریان‌ها و گوش دادن به به‌روزرسانی‌های پایگاه داده 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 یا خارج از یک coroutine با تابع trySend منتشر شوند.

در داخل، callbackFlow از یک کانال استفاده می‌کند که از نظر مفهومی بسیار شبیه به یک صف مسدودکننده است. یک کانال با ظرفیت ، حداکثر تعداد عناصری که می‌توانند بافر شوند، پیکربندی شده است. کانال ایجاد شده در callbackFlow دارای ظرفیت پیش‌فرض ۶۴ عنصر است. وقتی سعی می‌کنید یک عنصر جدید را به یک کانال کامل اضافه کنید، send تولیدکننده را تا زمانی که فضای کافی برای عنصر جدید وجود داشته باشد، به حالت تعلیق در می‌آورد، در حالی که trySend عنصر را به کانال اضافه نمی‌کند و بلافاصله false را برمی‌گرداند.

trySend بلافاصله عنصر مشخص شده را به کانال اضافه می‌کند، تنها در صورتی که این کار محدودیت‌های ظرفیت آن را نقض نکند، و سپس نتیجه موفقیت‌آمیز را برمی‌گرداند.

منابع جریان اضافی

• تست جریان‌های کاتلین روی اندروید
• StateFlow و SharedFlow
• منابع اضافی برای کوروتین‌ها و جریان کاتلین