কোটলিন অ্যান্ড্রয়েডে প্রবাহিত হয়

কো-রুটিনে, ফ্লো হলো এমন এক ধরনের টাইপ যা ক্রমানুসারে একাধিক ভ্যালু নির্গত করতে পারে; এর বিপরীতে সাসপেন্ড ফাংশন কেবল একটিমাত্র ভ্যালু রিটার্ন করে। উদাহরণস্বরূপ, আপনি ডাটাবেস থেকে লাইভ আপডেট পাওয়ার জন্য একটি ফ্লো ব্যবহার করতে পারেন।

ফ্লো কো-রুটিনের উপর ভিত্তি করে তৈরি এবং এটি একাধিক মান প্রদান করতে পারে। ধারণাগতভাবে, ফ্লো হলো ডেটার একটি প্রবাহ যা অ্যাসিঙ্ক্রোনাসভাবে গণনা করা যায়। নির্গত মানগুলো অবশ্যই একই ধরনের হতে হবে। উদাহরণস্বরূপ, একটি Flow<Int> হলো এমন একটি ফ্লো যা পূর্ণসংখ্যার মান নির্গত করে।

একটি ফ্লো অনেকটা Iterator মতোই, যা ধারাবাহিকভাবে বিভিন্ন ভ্যালু তৈরি করে, কিন্তু এটি অ্যাসিঙ্ক্রোনাসভাবে ভ্যালু তৈরি ও গ্রহণ করার জন্য সাসপেন্ড ফাংশন ব্যবহার করে। এর মানে হলো, উদাহরণস্বরূপ, ফ্লোটি মেইন থ্রেডকে ব্লক না করেই পরবর্তী ভ্যালুটি তৈরি করার জন্য নিরাপদে একটি নেটওয়ার্ক রিকোয়েস্ট করতে পারে।

তথ্য প্রবাহের সাথে তিনটি সত্তা জড়িত থাকে:

• একজন প্রডিউসার ডেটা তৈরি করে যা স্ট্রিমে যুক্ত হয়। কো-রুটিনের কল্যাণে, ফ্লো অ্যাসিঙ্ক্রোনাসভাবেও ডেটা তৈরি করতে পারে।
• (ঐচ্ছিক) মধ্যস্থতাকারীরা স্ট্রিমে নির্গত প্রতিটি মান অথবা স্বয়ং স্ট্রিমটিকেই পরিবর্তন করতে পারে।
• একজন গ্রাহক স্ট্রিম থেকে মানগুলো গ্রহণ করে।

অ্যান্ড্রয়েডে, একটি রিপোজিটরি সাধারণত UI ডেটার উৎপাদক হিসেবে কাজ করে, যার ভোক্তা হলো ইউজার ইন্টারফেস (UI), যা শেষ পর্যন্ত সেই ডেটা প্রদর্শন করে। আবার কখনও কখনও, UI লেয়ারটি ইউজার ইনপুট ইভেন্টের উৎপাদক হয় এবং হায়ারার্কির অন্যান্য লেয়ারগুলো সেগুলো গ্রহণ করে। উৎপাদক এবং ভোক্তার মধ্যবর্তী লেয়ারগুলো সাধারণত মধ্যস্থতাকারী হিসেবে কাজ করে, যা ডেটার প্রবাহকে পরিবর্তন করে পরবর্তী লেয়ারের প্রয়োজন অনুযায়ী সামঞ্জস্য করে।

একটি প্রবাহ তৈরি করা

ফ্লো তৈরি করতে ফ্লো বিল্ডার এপিআই ব্যবহার করুন। 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 বিল্ডার ব্যবহার করলে, প্রডিউসার অন্য কোনো 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) }
}
Flow.kt

মধ্যবর্তী অপারেটরগুলো একের পর এক প্রয়োগ করা যেতে পারে, যা এমন একটি অপারেশন শৃঙ্খল তৈরি করে যা ফ্লো-তে কোনো আইটেম নির্গত হলে অলসভাবে (lazily) সম্পাদিত হয়। উল্লেখ্য যে, কোনো স্ট্রিমে শুধু একটি মধ্যবর্তী অপারেটর প্রয়োগ করলেই ফ্লো কালেকশন শুরু হয় না।

একটি প্রবাহ থেকে সংগ্রহ করা

ভ্যালু শোনা শুরু করার জন্য ফ্লো-কে ট্রিগার করতে একটি টার্মিনাল অপারেটর ব্যবহার করুন। স্ট্রিমে নির্গত হওয়ার সাথে সাথে সমস্ত ভ্যালু পেতে, collect ব্যবহার করুন। আপনি অফিসিয়াল ফ্লো ডকুমেন্টেশনে টার্মিনাল অপারেটর সম্পর্কে আরও জানতে পারবেন।

যেহেতু collect একটি সাসপেন্ড ফাংশন, তাই এটিকে একটি কো-রুটিনের মধ্যে এক্সিকিউট করতে হবে। এটি প্যারামিটার হিসেবে একটি ল্যাম্বডা গ্রহণ করে, যা প্রতিটি নতুন ভ্যালুর জন্য কল করা হয়। যেহেতু এটি একটি সাসপেন্ড ফাংশন, তাই যে কো-রুটিনটি 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 ল্যাম্বডাটি কল করা হয়, কারণ ব্যতিক্রমটির কারণে স্ট্রিমে একটি নতুন আইটেম নির্গত হয়েছে।

ভিন্ন একটি CoroutineContext-এ কার্যকর করা হচ্ছে

ডিফল্টরূপে, একটি flow বিল্ডারের প্রডিউসার সেই CoroutineContext এর মধ্যেই এক্সিকিউট হয় যা এটি থেকে ডেটা সংগ্রহ করে, এবং পূর্বেই যেমন উল্লেখ করা হয়েছে, এটি অন্য কোনো CoroutineContext থেকে ভ্যালু emit করতে পারে না। এই আচরণটি কিছু ক্ষেত্রে অনাকাঙ্ক্ষিত হতে পারে। উদাহরণস্বরূপ, এই টপিক জুড়ে ব্যবহৃত উদাহরণগুলিতে, রিপোজিটরি লেয়ারের সেই Dispatchers.Main এর উপর কোনো অপারেশন করা উচিত নয় যা viewModelScope দ্বারা ব্যবহৃত হয়।

একটি ফ্লো-এর CoroutineContext পরিবর্তন করতে, flowOn ইন্টারমিডিয়েট অপারেটরটি ব্যবহার করুন। flowOn আপস্ট্রিম ফ্লো- এর CoroutineContext পরিবর্তন করে, অর্থাৎ প্রডিউসার এবং flowOn আগে (বা উপরে) প্রয়োগ করা যেকোনো ইন্টারমিডিয়েট অপারেটরের CoroutineContext পরিবর্তন করে। ডাউনস্ট্রিম ফ্লো ( flowOn পরের ইন্টারমিডিয়েট অপারেটরগুলো এবং কনজিউমার) প্রভাবিত হয় না এবং ফ্লো থেকে collect জন্য ব্যবহৃত CoroutineContext উপরই এক্সিকিউট হয়। যদি একাধিক 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 এক্সিকিউট হয় 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)
}
Flow.kt

জেটপ্যাক লাইব্রেরিতে প্রবাহ

ফ্লো অনেক জেটপ্যাক লাইব্রেরির সাথে সমন্বিত, এবং এটি অ্যান্ড্রয়েড থার্ড-পার্টি লাইব্রেরিগুলোর মধ্যে জনপ্রিয়। লাইভ ডেটা আপডেট এবং অফুরন্ত ডেটা প্রবাহের জন্য ফ্লো একটি চমৎকার সমাধান।

ডাটাবেসের পরিবর্তন সম্পর্কে অবহিত হতে আপনি Room-এর সাথে Flow ব্যবহার করতে পারেন। ডেটা অ্যাক্সেস অবজেক্ট (DAO) ব্যবহার করার সময়, লাইভ আপডেট পেতে একটি Flow টাইপ রিটার্ন করুন।

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

যখনই Example টেবিলে কোনো পরিবর্তন হয়, ডাটাবেসের নতুন আইটেমগুলোসহ একটি নতুন তালিকা প্রকাশিত হয়।

কলব্যাক-ভিত্তিক এপিআইগুলোকে ফ্লো-তে রূপান্তর করুন

callbackFlow হলো একটি ফ্লো বিল্ডার যা আপনাকে কলব্যাক-ভিত্তিক এপিআইগুলোকে ফ্লো-তে রূপান্তর করতে দেয়। উদাহরণস্বরূপ, Firebase Firestore অ্যান্ড্রয়েড এপিআইগুলো কলব্যাক ব্যবহার করে।

এই API-গুলোকে ফ্লো-তে রূপান্তর করতে এবং ফায়ারস্টোর ডাটাবেস আপডেটের জন্য লিসেন করতে, আপনি নিম্নলিখিত কোডটি ব্যবহার করতে পারেন:

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 একটি চ্যানেল ব্যবহার করে, যা ধারণাগতভাবে একটি ব্লকিং কিউ- এর (blocking queue) মতোই। একটি চ্যানেলকে তার ধারণক্ষমতা (capacity) দিয়ে কনফিগার করা হয়, যা হলো বাফার করা যেতে পারে এমন উপাদানের সর্বোচ্চ সংখ্যা। callbackFlow তে তৈরি করা চ্যানেলটির ডিফল্ট ধারণক্ষমতা হলো ৬৪টি উপাদান। যখন আপনি একটি পূর্ণ চ্যানেলে নতুন উপাদান যোগ করার চেষ্টা করেন, send প্রডিউসারকে (producer) স্থগিত করে রাখে যতক্ষণ না নতুন উপাদানটির জন্য জায়গা তৈরি হয়, অন্যদিকে trySend উপাদানটিকে চ্যানেলে যোগ করে না এবং সাথে সাথে false রিটার্ন করে।

trySend নির্দিষ্ট উপাদানটিকে অবিলম্বে চ্যানেলে যুক্ত করে, তবে শুধুমাত্র যদি এটি চ্যানেলের ধারণক্ষমতার সীমাবদ্ধতা লঙ্ঘন না করে, এবং তারপর সফলতার ফলাফল ফেরত দেয়।

অতিরিক্ত প্রবাহ সম্পদ

• অ্যান্ড্রয়েডে কোটলিন ফ্লো পরীক্ষা করা
• StateFlow এবং SharedFlow
• কোটলিন কোরাউটিন এবং ফ্লো সম্পর্কিত অতিরিক্ত রিসোর্স