Flow — چیست، استریم‌های cold و hot در کوروتین‌های Kotlin

Flow در Kotlin چیست؟

Flow — یک نوع از بسته kotlinx.coroutines.flow است که یک جریان داده ناهم‌زمان cold را نشان می‌دهد. در اصل Flow یک دنباله کوروتینی است که مقادیر را از طریق تابع emit() منتشر می‌کند و یا با موفقیت یا با یک استثنا پایان می‌یابد. جمع‌آوری جریان از طریق عملگر terminal collect() انجام می‌شود که یک تابع suspend است.

معناشناسی cold

Cold stream به این معنی است که کد داخل flow-builder برای هر مشترک دوباره اجرا می‌شود. Observable.fromIterable در RxJava رفتار مشابهی دارد: مشترک جدید همه مقادیر را از ابتدا دریافت می‌کند. در Flow این از طریق تابع suspend collect پیاده‌سازی شده است که کوروتین را در تمام مدت جمع‌آوری داده مسدود می‌کند.

سازنده‌های Flow

Kotlin چندین روش برای ایجاد Flow فراهم می‌کند: flow { } — ساختار پایه با emit()، flowOf(vararg values) — برای مجموعه ثابتی از مقادیر، .asFlow() — extension برای مجموعه‌ها و Sequence. همه سازنده‌ها cold هستند — داده‌ها فقط هنگام فراخوانی عملگر terminal تولید می‌شوند.

استریم‌های Cold و Hot

تقسیم به استریم‌های cold و hot یک مفهوم کلیدی در برنامه‌نویسی reactive است. Cold stream (Flow, Observable) تولید داده را هنگام اشتراک آغاز می‌کند. Hot stream (Channel, SharedFlow) داده را مستقل منتشر می‌کند — مشترک فقط آنچه را که پس از اشتراک رخ می‌دهد دریافت می‌کند، بدون شروع دنباله.

SharedFlow — یک Flow hot است که می‌تواند مشترکین متعددی داشته باشد و با تنظیم replay می‌تواند آخرین مقادیر را دوباره پخش کند. SharedFlow برای رویدادها مناسب است (اعلان‌های یکبار مصرف). StateFlow — نوعی از آن با یک مقدار حالت ثابت است که آخرین مقدار را برای مشترکین جدید ذخیره می‌کند.

ChannelFlow از Channel در زیرساخت استفاده می‌کند و ویژگی‌های Flow و Channel را ترکیب می‌کند. از بافرینگ و backpressure از طریق ظرفیت (capacity) پشتیبانی می‌کند. ChannelFlow هنگام تبدیل callback-API به جریان reactive مفید است، زمانی که مقادیر از کوروتین‌های مختلف منتشر می‌شوند.

تبدیل بین cold و hot

برای تبدیل cold Flow به hot SharedFlow از عملگر shareIn(scope, started, replay) استفاده می‌شود. پارامتر started لحظه شروع را کنترل می‌کند: SharingStarted.WhileSubscribed() — فعال تا زمانی که مشترک وجود دارد، Lazily — شروع با اولین مشترک، Eagerly — شروع فوری. تبدیل معکوس — hot به cold: StateFlow.asFlow() یک Flow cold برمی‌گرداند که هنگام collect مقدار فعلی StateFlow را منتشر می‌کند. این برای تست مناسب است.

عملگرهای Flow

Flow مجموعه غنی از عملگرها را فراهم می‌کند که به عنوان توابع suspend در داخل کوروتین کار می‌کنند. عملگرها حالت ندارند و یک Flow جدید برمی‌گردانند — جریان اصلی بدون تغییر باقی می‌ماند. این امکان ساخت زنجیره‌های تبدیل امن بدون عوارض جانبی را فراهم می‌کند.

عملگر map هر مقدار جریان را از طریق تبدیل ناهم‌زمان یا هم‌زمان تغییر می‌دهد. filter فقط مقادیری را که شرط را برآورده می‌کنند عبور می‌دهد. catch استثناها را قبل از عملگر terminal می‌گیرد و امکان بازیابی جریان را فراهم می‌کند. flatMapLatest انتشار قبلی را هنگام دریافت مقدار جدید لغو می‌کند — مشابه switchMap در Rx.

عملگر debounce در Flow انتشار مقدار را برای مدت زمان مشخصی به تأخیر می‌اندازد. اگر در این مدت مقدار جدیدی بیاید — تایمر بازنشانی می‌شود. در Android از debounce برای جستجو استفاده می‌شود: درخواست فقط پس از مکث 300-400 میلی‌ثانیه ارسال می‌شود که تعداد فراخوانی‌های API را 3-5 برابر کاهش می‌دهد.

عملگرهای terminal

علاوه بر collect()، Flow از عملگرهای terminal دیگری نیز پشتیبانی می‌کند: toList() همه مقادیر را در یک لیست جمع‌آوری می‌کند — برای تست مفید است، first() اولین عنصر را برمی‌گرداند و جریان را لغو می‌کند، single() دقیقاً یک عنصر را انتظار دارد. fold(initial) مقادیر را از طریق تابع داده شده جمع می‌کند. همه عملگرهای terminal توابع suspend هستند و باید در داخل کوروتین یا تابع suspend دیگر فراخوانی شوند.

نمونه کد Flow

مثال اول — Flow پایه با تولید اعداد و تبدیل از طریق عملگر map:

val numberFlow = flow {
    for (i in 1..5) {
        delay(500)
        emit(i)
    }
}

scope.launch {
    numberFlow
        .map { "عدد: $it" }
        .collect { value ->
            println(value)
        }
}

مثال دوم — تبدیل جریان با فیلتر کردن و مدیریت خطا از طریق catch:

flow {
    emit("data1")
    emit("data2")
    throw RuntimeException("network error")
}
    .catch { e ->
        emit("fallback_data")
    }
    .collect { value ->
        println(value)
    }

مثال سوم — استفاده از StateFlow در ViewModel برای UI reactive در Jetpack Compose:

class SearchViewModel : ViewModel() {
    private val _query = MutableStateFlow("")
    val results: StateFlow<List<Result>> = _query
        .debounce(300)
        .flatMapLatest { query ->
            repository.search(query)
        }
        .catch { emit(emptyList()) }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), emptyList())

    fun onQueryChanged(query: String) {
        _query.value = query
    }
}

StateFlow و SharedFlow

StateFlow — یک Flow hot با یک مقدار جاری واحد است. آخرین مقدار را ذخیره می‌کند و آن را فوراً به مشترک جدید منتقل می‌کند. StateFlow یک کانتینر Observable برای حالت است، از مقایسه equals پشتیبانی می‌کند — اگر مقدار جدید با مقدار فعلی مطابقت داشته باشد، انتشار رخ نمی‌دهد. Jetpack Compose از StateFlow از طریق collectAsState() استفاده می‌کند.

SharedFlow — یک Flow hot انعطاف‌پذیرتر بدون مقدار اولیه اجباری است. SharedFlow از طریق replay (تعداد مقادیر برای مشترکین جدید)، extraBufferCapacity (بافر فراتر از replay) و onBufferOverflow (استراتژی هنگام سرریز) پیکربندی می‌شود. SharedFlow برای رویدادهای یکبار مصرف ایده‌آل است: ناوبری، Snackbar، تحلیل.

Flow در معماری Android توسط Google به عنوان منبع اصلی داده توصیه می‌شود (لایه: Repository → UseCase → ViewModel). LiveData از نظر انعطاف‌پذیری از Flow پایین‌تر است: Flow از کوروتین‌ها، عملگرها، backpressure پشتیبانی می‌کند و در خارج از لایه UI کار می‌کند. مهاجرت از LiveData به Flow یک روش استاندارد در پروژه‌های مدرن Android است.

هنگام استفاده از Flow در ViewModel انتخاب نوع صحیح مهم است. StateFlow برای حالت UI که باید چرخش صفحه را تحمل کند ایده‌آل است. SharedFlow برای رویدادهایی که پردازش مجدد مجاز نیست مناسب است — مثلاً ناوبری. Flow با collect() در lifecycleScope حداکثر کنترل را بر زمینه اجرا می‌دهد، اما هنگام خروج از صفحه نیاز به لغو دستی دارد.

تست Flow از طریق kotlinx-coroutines-test انجام می‌شود. کتابخانه TestDispatcher — زمان مجازی را فراهم می‌کند که امکان تسریع تأخیرها (delay) و کنترل ترتیب اجرای کوروتین‌ها را می‌دهد. TestScope.runTest { } یک محیط ایزوله برای تست Flow ایجاد می‌کند. عملگر toList() اغلب در تست‌ها برای جمع‌آوری همه مقادیر flow با timeout استفاده می‌شود تا بررسی شود که جریان دنباله صحیح داده را منتشر کرده است.

Flow به خوبی با Room (کتابخانه پایگاه داده Android) یکپارچه می‌شود: متدهای DAO می‌توانند Flow<List<Entity>> برگردانند. Room به طور خودکار با هر تغییری در جدول مقدار جدیدی منتشر می‌کند — UI بدون تریگر دستی به‌روزرسانی می‌شود. این از طریق InvalidationTracker پیاده‌سازی شده است که در زیرساخت از Flow با callbackFlow استفاده می‌کند. چنین رویکردی نیاز به LiveData را از بین می‌برد و لایه داده را کاملاً کوروتین‌محور می‌کند. Jetpack Compose از طریق collectAsState() در StateFlow مشترک می‌شود و فقط مؤلفه‌هایی را که داده‌هایشان تغییر کرده دوباره ترسیم می‌کند — این عملکردی را فراهم می‌کند که با معماری‌های LiveData محور دست‌یافتنی نیست. DataStore (جایگزین SharedPreferences) نیز Flow<Preferences> را برمی‌گرداند و خواندن reactive تنظیمات برنامه را بدون تریگرهای دستی به‌روزرسانی تضمین می‌کند.

Flow از ارتباط بین فرآیندی از طریق kotlinx-coroutines-core روی JVM بدون کتابخانه‌های اضافی پشتیبانی می‌کند. مثلاً در برنامه‌های سروری روی Ktor، Flow می‌تواند جریان پیام‌های WebSocket ورودی را نشان دهد. هر پیام به جریان منتشر می‌شود، از فیلتر و تجمیع از طریق عملگرها عبور می‌کند و نتیجه به کلاینت ارسال می‌شود. چنین رویکردی جایگزین کتابخانه‌های reactive مانند Reactor یا RxJava در پروژه‌های Kotlin می‌شود.

سازگاری Flow با کد RxJava موجود توسط ماژول kotlinx-coroutines-rx3 تضمین می‌شود. تابع extension Flow.asObservable() Flow را به Observable از RxJava 3 تبدیل می‌کند. تبدیل معکوس — CompletableSource.asFlow(), Observable.asFlow(). این مهاجرت از RxJava به کوروتین‌ها را ساده می‌کند: می‌توان پروژه را مرحله‌ای بازنویسی کرد و بخشی از لایه‌ها را روی RxJava باقی گذاشت. هنگام تبدیل باید تفاوت در معناشناسی cold/hot را در نظر گرفت: Observable می‌تواند هم cold و هم hot باشد، Flow همیشه cold برای Flow معمولی و hot برای SharedFlow است.

مدیریت خطا و تست Flow

برای مدیریت خطا در Flow یک ویژگی وجود دارد: اگر استثنا در داخل flow-builder قبل از عملگر terminal رخ دهد، به catch منتقل می‌شود. اگر استثنا در عملگر بعد از builder رخ دهد، catch بعد از آن عملگر آن را می‌گیرد. retryWhen امکان تکرار اشتراک با شرط را فراهم می‌کند: در خطای شبکه تا 3 بار تکرار کن، اما در CancellationException تکرار نکن. Flow خطاهای وابسته به حالت را حذف می‌کند، زیرا حالت را ذخیره نمی‌کند — این کار debug را در مقایسه با Observable که در آن Subject حالت داخلی را ذخیره می‌کند ساده‌تر می‌کند.

تست Flow با kotlinx-coroutines-test از TestDispatcher برای شبیه‌سازی تأخیرها استفاده می‌کند. Turbine — کتابخانه محبوب جامعه برای تست Flow: test { } Flow را راه‌اندازی می‌کند، awaitItem() منتظر مقدار بعدی می‌ماند، awaitComplete() منتظر اتمام می‌ماند. Turbine یک timeout پیش‌فرض اضافه می‌کند که از هنگ کردن تست‌ها جلوگیری می‌کند. برای تست StateFlow از .testIn(scope) با بررسی مقادیر به ترتیب زمانی استفاده کنید.