Android पर Kotlin फ़्लो

कोरूटीन में, फ़्लो एक ऐसा टाइप है जो क्रम से कई वैल्यू दिखा सकता है. वहीं, सस्पेंड फ़ंक्शन सिर्फ़ एक वैल्यू दिखाते हैं. उदाहरण के लिए, डेटाबेस से लाइव अपडेट पाने के लिए, फ़्लो का इस्तेमाल किया जा सकता है.

फ़्लो, कोरूटीन पर बनाए जाते हैं और कई वैल्यू दिखा सकते हैं. फ़्लो, कॉन्सेप्ट के तौर पर डेटा की स्ट्रीम होती है, जिसे एसिंक्रोनस तरीके से कैलकुलेट किया जा सकता है. दिखाई गई वैल्यू, एक ही टाइप की होनी चाहिए. उदाहरण के लिए, 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 बिल्डर, कोरूटीन में एक्ज़ीक्यूट होता है. इसलिए, यह एसिंक्रोनस एपीआई से फ़ायदा लेता है. हालांकि, इस पर कुछ पाबंदियां लागू होती हैं:

  • फ़्लो सीक्वेंशियल होते हैं. प्रड्यूसर, कोरूटीन में होता है. इसलिए, सस्पेंड फ़ंक्शन को कॉल करने पर, प्रड्यूसर तब तक सस्पेंड रहता है, जब तक सस्पेंड फ़ंक्शन कोई वैल्यू नहीं दिखाता. उदाहरण के लिए, प्रड्यूसर तब तक सस्पेंड रहता है, जब तक fetchLatestNews नेटवर्क अनुरोध पूरा नहीं हो जाता. इसके बाद ही, नतीजे को स्ट्रीम में दिखाया जाता है.
  • flow बिल्डर की मदद से, प्रड्यूसर किसी दूसरे CoroutineContext से वैल्यू emit नहीं कर सकता. इसलिए, नई कोरूटीन बनाकर या withContext कोड ब्लॉक का इस्तेमाल करके, किसी दूसरे CoroutineContext में emit को कॉल न करें. इन मामलों में, callbackFlow जैसे अन्य फ़्लो बिल्डर का इस्तेमाल किया जा सकता है.

स्ट्रीम में बदलाव करना

इंटरमीडियरी, वैल्यू का इस्तेमाल किए बिना, डेटा की स्ट्रीम में बदलाव करने के लिए इंटरमीडिएट ऑपरेटर का इस्तेमाल कर सकते हैं. ये ऑपरेटर ऐसे फ़ंक्शन होते हैं जो डेटा की स्ट्रीम पर लागू होने पर, कार्रवाइयों की एक चेन सेट अप करते हैं. ये चेन तब तक एक्ज़ीक्यूट नहीं होतीं, जब तक वैल्यू का इस्तेमाल नहीं किया जाता. फ़्लो के रेफ़रंस दस्तावेज़ में, इंटरमीडिएट ऑपरेटर के बारे में ज़्यादा जानें .

यहां दिए गए उदाहरण में, रिपॉज़िटरी लेयर, View पर दिखने वाले डेटा को बदलने के लिए, इंटरमीडिएट ऑपरेटर map का इस्तेमाल करती है:

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 एक सस्पेंड फ़ंक्शन है. इसलिए, इसे कोरूटीन में एक्ज़ीक्यूट करना ज़रूरी है. यह पैरामीटर के तौर पर एक लैम्डा लेता है, जिसे हर नई वैल्यू पर कॉल किया जाता है. यह एक सस्पेंड फ़ंक्शन है. इसलिए, collect को कॉल करने वाली कोरूटीन, फ़्लो बंद होने तक सस्पेंड हो सकती है.

पिछले उदाहरण को जारी रखते हुए, यहां ViewModel को लागू करने का एक आसान तरीका बताया गया है. यह 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 लैम्डा को कॉल नहीं किया जाता, क्योंकि कोई नया आइटम नहीं मिला है.

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 लैम्डा को कॉल किया जाता है, क्योंकि अपवाद की वजह से स्ट्रीम में एक नया आइटम दिखाया गया है.

किसी दूसरे 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())
            }
}

इस कोड की मदद से, onEach और map ऑपरेटर, defaultDispatcher का इस्तेमाल करते हैं. वहीं, catch ऑपरेटर और कंज्यूमर, viewModelScope से इस्तेमाल किए जाने वाले Dispatchers.Main पर एक्ज़ीक्यूट होते हैं.

डेटा सोर्स लेयर, I/O का काम कर रही है. इसलिए, आपको ऐसे डिस्पैचर का इस्तेमाल करना चाहिए जिसे I/O कार्रवाइयों के लिए ऑप्टिमाइज़ किया गया हो:

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

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

Jetpack लाइब्रेरी में फ़्लो

फ़्लो को कई Jetpack लाइब्रेरी में इंटिग्रेट किया गया है. साथ ही, यह Android की तीसरे पक्ष की लाइब्रेरी में भी लोकप्रिय है. फ़्लो, लाइव डेटा अपडेट और डेटा की कभी न खत्म होने वाली स्ट्रीम के लिए सबसे सही है.

डेटाबेस में हुए बदलावों के बारे में सूचना पाने के लिए, Room के साथ फ़्लो का इस्तेमाल किया जा सकता है. डेटा ऐक्सेस ऑब्जेक्ट (डीएओ) का इस्तेमाल करते समय, लाइव अपडेट पाने के लिए Flow टाइप दिखाएं.

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

Example टेबल में हर बार बदलाव होने पर, डेटाबेस में मौजूद नए आइटम के साथ एक नई सूची दिखाई जाती है.

कॉलबैक पर आधारित एपीआई को फ़्लो में बदलना

callbackFlow एक फ़्लो बिल्डर है. इसकी मदद से, कॉल बैक पर आधारित एपीआई को फ़्लो में बदला जा सकता है. उदाहरण के लिए, Firebase Firestore Android एपीआई, कॉल बैक का इस्तेमाल करते हैं.

इन एपीआई को फ़्लो में बदलने और 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 की मदद से, send फ़ंक्शन का इस्तेमाल करके, किसी दूसरे CoroutineContext से या trySend फ़ंक्शन का इस्तेमाल करके, कोरूटीन के बाहर से वैल्यू दिखाई जा सकती हैं.

इंटरनली, callbackFlow एक चैनल का इस्तेमाल करता है. यह कॉन्सेप्ट के तौर पर, ब्लॉकिंग क्यू से काफ़ी मिलता-जुलता है. किसी चैनल को कैपेसिटी के साथ कॉन्फ़िगर किया जाता है. यह बफ़र किए जा सकने वाले एलिमेंट की ज़्यादा से ज़्यादा संख्या होती है . callbackFlow में बनाए गए चैनल की डिफ़ॉल्ट कैपेसिटी 64 एलिमेंट होती है. जब किसी भरे हुए चैनल में कोई नया एलिमेंट जोड़ने की कोशिश की जाती है, तो send, प्रड्यूसर को तब तक सस्पेंड रखता है, जब तक नए एलिमेंट के लिए जगह नहीं बन जाती. वहीं, trySend, चैनल में एलिमेंट नहीं जोड़ता और तुरंत false दिखाता है.

trySend , चैनल में तय किया गया एलिमेंट तुरंत जोड़ देता है. हालांकि, ऐसा सिर्फ़ तब होता है, जब इससे इसकी कैपेसिटी से जुड़ी पाबंदियों का उल्लंघन न हो. इसके बाद, यह नतीजे के तौर पर 'सफल' दिखाता है.

फ़्लो के अन्य संसाधन