Flow — یہ کیا ہے، Kotlin کوروٹینز میں cold اور hot اسٹریمز

Kotlin میں Flow کیا ہے؟

Flow kotlinx.coroutines.flow پیکیج کی ایک قسم ہے، جو ایک cold غیر مطابقت پذیر ڈیٹا اسٹریم کی نمائندگی کرتی ہے۔ اس کے جوہر میں، Flow ایک کوروٹین ترتیب ہے جو emit() فنکشن کے ذریعے اقدار جاری کرتی ہے اور کامیابی سے یا کسی استثنا کے ساتھ ختم ہوتی ہے۔ اسٹریم جمع کرنا ٹرمینل آپریٹر collect() کے ذریعے کیا جاتا ہے، جو ایک suspend فنکشن ہے۔

Cold سیمنٹکس

Cold اسٹریم کا مطلب ہے کہ flow بلڈر کے اندر کا کوڈ ہر سبسکرائبر کے لیے نئے سرے سے چلتا ہے۔ RxJava میں Observable.fromIterable اسی طرح برتاؤ کرتا ہے: ایک نیا سبسکرائبر شروع سے تمام اقدار حاصل کرتا ہے۔ Flow میں، یہ suspend فنکشن collect کے ذریعے لاگو کیا گیا ہے، جو ڈیٹا جمع کرنے کی پوری مدت کے لیے کوروٹین کو بلاک کرتا ہے۔

Flow بلڈرز

Kotlin Flow بنانے کے کئی طریقے فراہم کرتا ہے: flow { } — emit() کے ساتھ بنیادی تعمیر، flowOf(vararg values) — اقدار کے ایک مقررہ سیٹ کے لیے، .asFlow() — کلیکشنز اور Sequence کے لیے ایک توسیع۔ تمام بلڈرز cold ہیں — ڈیٹا صرف تب پیدا ہوتا ہے جب ٹرمینل آپریٹر کو کال کیا جائے۔

Cold اور Hot اسٹریمز

Cold اور hot اسٹریمز میں تقسیم ری ایکٹیو پروگرامنگ کا ایک کلیدی تصور ہے۔ Cold اسٹریم (Flow, Observable) سبسکرپشن پر ڈیٹا جنریشن شروع کرتا ہے۔ Hot اسٹریم (Channel, SharedFlow) آزادانہ طور پر ڈیٹا جاری کرتا ہے — سبسکرائبر صرف وہی حاصل کرتا ہے جو سبسکرپشن کے بعد ہوتا ہے، ترتیب کے آغاز کے بغیر۔

SharedFlow ایک hot Flow ہے جس میں متعدد سبسکرائبر ہو سکتے ہیں اور جب replay کنفیگر کیا جائے تو حالیہ اقدار کو دوبارہ چلا سکتا ہے۔ SharedFlow واقعات (ایک بار کی اطلاع) کے لیے موزوں ہے۔ StateFlow اس کی ایک قسم ہے جس میں ایک مقررہ حالت کی قدر ہے، جو نئے سبسکرائبرز کے لیے آخری قدر کو کیش کرتی ہے۔

ChannelFlow اندرونی طور پر Channel استعمال کرتا ہے، Flow اور Channel کی خصوصیات کو یکجا کرتا ہے۔ یہ capacity کے ذریعے بفرنگ اور بیک پریشر کو سپورٹ کرتا ہے۔ ChannelFlow کال بیک APIs کو ری ایکٹیو اسٹریم میں تبدیل کرتے وقت مفید ہے، جہاں اقدار مختلف کوروٹینز سے جاری کی جاتی ہیں۔

Cold اور hot کے درمیان تبدیلی

Cold Flow کو hot SharedFlow میں تبدیل کرنے کے لیے shareIn(scope, started, replay) آپریٹر استعمال کیا جاتا ہے۔ پیرامیٹر started شروع ہونے کے لمحے کو کنٹرول کرتا ہے: SharingStarted.WhileSubscribed() — جب تک سبسکرائبر ہیں فعال، Lazily — پہلے سبسکرائبر پر شروع، Eagerly — فوری شروع۔ الٹی تبدیلی — hot سے cold: StateFlow.asFlow() ایک cold Flow لوٹاتا ہے جو collect پر StateFlow کی موجودہ قدر جاری کرتا ہے۔ یہ جانچ کے لیے آسان ہے۔

Flow آپریٹرز

Flow آپریٹرز کا ایک بھرپور سیٹ فراہم کرتا ہے جو کوروٹین کے اندر suspend فنکشنز کے طور پر کام کرتے ہیں۔ آپریٹرز اسٹیٹ لیس ہوتے ہیں اور ایک نیا Flow لوٹاتے ہیں — اصل اسٹریم تبدیل نہیں ہوتا۔ یہ ضمنی اثرات کے بغیر محفوظ تبدیلی کی زنجیریں بنانے کی اجازت دیتا ہے۔

map آپریٹر ہر اسٹریم قدر کو ایک غیر مطابقت پذیر یا مطابقت پذیر تبدیلی کے ذریعے تبدیل کرتا ہے۔ filter صرف ان اقدار کو گزرنے دیتا ہے جو شرط کو پورا کرتی ہیں۔ catch ٹرمینل آپریٹر سے پہلے استثناؤں کو پکڑتا ہے اور اسٹریم کی بحالی کی اجازت دیتا ہے۔ flatMapLatest جب نئی قدر آتی ہے تو پچھلے اجراء کو منسوخ کرتا ہے — Rx میں switchMap کی طرح۔

Flow میں debounce آپریٹر قدر کی اشاعت کو ایک مخصوص ٹائم آؤٹ تک موخر کرتا ہے۔ اگر اس دوران کوئی نئی قدر آتی ہے، تو ٹائمر دوبارہ سیٹ ہو جاتا ہے۔ Android میں، debounce تلاش کے لیے استعمال ہوتا ہے: درخواست صرف 300-400 ms کے وقفے کے بعد بھیجی جاتی ہے، جس سے API کالز 3-5 گنا کم ہو جاتی ہیں۔

ٹرمینل آپریٹرز

collect() کے علاوہ، Flow دیگر ٹرمینل آپریٹرز کو سپورٹ کرتا ہے: toList() تمام اقدار کو ایک فہرست میں جمع کرتا ہے — جانچ کے لیے مفید، first() پہلا عنصر لوٹاتا ہے اور اسٹریم کو منسوخ کرتا ہے، single() بالکل ایک عنصر کی توقع کرتا ہے۔ fold(initial) ایک منتقل کردہ فنکشن کے ذریعے اقدار کو جمع کرتا ہے۔ تمام ٹرمینل آپریٹرز suspend فنکشنز ہیں اور انہیں کوروٹین یا کسی دوسرے suspend فنکشن کے اندر کال کیا جانا چاہیے۔

Flow کوڈ کی مثالیں

پہلی مثال — map آپریٹر کے ذریعے تبدیلی کے ساتھ نمبر پیدا کرنے والا بنیادی Flow:

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)
    }

تیسری مثال — Jetpack Compose میں ری ایکٹیو UI کے لیے ViewModel میں StateFlow کا استعمال:

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 ایک واحد موجودہ قدر کے ساتھ ایک hot Flow ہے۔ یہ آخری قدر کو کیش کرتا ہے اور اسے فوری طور پر ایک نئے سبسکرائبر کو منتقل کرتا ہے۔ StateFlow حالت کے لیے ایک قابل مشاہدہ کنٹینر ہے، equals موازنہ کو سپورٹ کرتا ہے — اگر نئی قدر موجودہ سے مماثل ہے تو کوئی اجراء نہیں ہوتا۔ Jetpack Compose collectAsState() کے ذریعے StateFlow استعمال کرتا ہے۔

SharedFlow لازمی ابتدائی قدر کے بغیر ایک زیادہ لچکدار hot Flow ہے۔ SharedFlow کو replay (نئے سبسکرائبرز کے لیے اقدار کی تعداد)، extraBufferCapacity (replay سے باہر بفر)، اور onBufferOverflow (اوور فلو پر حکمت عملی) کے ذریعے کنفیگر کیا جاتا ہے۔ SharedFlow ایک بار کے واقعات کے لیے مثالی ہے: نیویگیشن، Snackbar، تجزیات۔

Android آرکیٹیکچر میں Flow کو Google کے ذریعہ بنیادی ڈیٹا سورس (پرت: Repository → UseCase → ViewModel) کے طور پر تجویز کیا گیا ہے۔ LiveData لچک میں Flow سے کمتر ہے: Flow کوروٹینز، آپریٹرز، بیک پریشر کو سپورٹ کرتا ہے اور UI پرت کے باہر کام کرتا ہے۔ LiveData سے Flow میں منتقلی جدید Android پروجیکٹس میں ایک معیاری عمل ہے۔

ViewModel میں Flow استعمال کرتے وقت صحیح قسم کا انتخاب کرنا ضروری ہے۔ StateFlow UI حالت کے لیے مثالی ہے جسے اسکرین گردش سے بچنا چاہیے۔ SharedFlow ان واقعات کے لیے موزوں ہے جہاں دوبارہ پروسیسنگ ناقابل قبول ہے — مثال کے طور پر، نیویگیشن۔ lifecycleScope میں collect() کے ساتھ Flow عملدرآمد کے سیاق و سباق پر زیادہ سے زیادہ کنٹرول دیتا ہے لیکن اسکرین چھوڑنے پر دستی منسوخی کی ضرورت ہوتی ہے۔

Flow کی جانچ kotlinx-coroutines-test کے ذریعے کی جاتی ہے۔ لائبریری TestDispatcher فراہم کرتی ہے — مجازی وقت جو تاخیر (delay) کو تیز کرنے اور کوروٹینز کے عملدرآمد کی ترتیب کو کنٹرول کرنے کی اجازت دیتا ہے۔ TestScope.runTest { } Flow کی جانچ کے لیے ایک الگ تھلگ ماحول بناتا ہے۔ toList() آپریٹر اکثر جانچوں میں ایک ٹائم آؤٹ کے ساتھ تمام flow اقدار کو جمع کرنے کے لیے استعمال ہوتا ہے، تاکہ تصدیق کی جا سکے کہ اسٹریم نے ڈیٹا کی صحیح ترتیب جاری کی۔

Flow Room (Android DB لائبریری) کے ساتھ اچھی طرح ضم ہوتا ہے: DAO کے طریقے Flow<List<Entity>> لوٹا سکتے ہیں۔ Room کسی بھی ٹیبل تبدیلی پر خود بخود ایک نئی قدر جاری کرتا ہے — UI دستی محرک کے بغیر اپ ڈیٹ ہوتا ہے۔ یہ InvalidationTracker کے ذریعے لاگو کیا گیا ہے، جو پردے کے پیچھے callbackFlow کے ساتھ Flow استعمال کرتا ہے۔ یہ نقطہ نظر LiveData کی ضرورت کو ختم کرتا ہے اور ڈیٹا پرت کو مکمل طور پر کوروٹین پر مبنی بناتا ہے۔ Jetpack Compose collectAsState() کے ذریعے StateFlow کو سبسکرائب کرتا ہے اور صرف ان اجزاء کو دوبارہ کھینچتا ہے جن کا ڈیٹا تبدیل ہوا ہے — یہ LiveData پر مبنی آرکیٹیکچرز کے ساتھ ناقابل حصول کارکردگی دیتا ہے۔ DataStore (SharedPreferences کا متبادل) بھی Flow<Preferences> لوٹاتا ہے، دستی اپ ڈیٹ محرکات کے بغیر ایپلیکیشن کی ترتیبات کا ری ایکٹیو پڑھنا فراہم کرتا ہے۔

Flow اضافی لائبریریوں کے بغیر JVM پر kotlinx-coroutines-core کے ذریعے انٹر پروسیس مواصلات کو سپورٹ کرتا ہے۔ مثال کے طور پر، Ktor پر سرور ایپلیکیشنز میں، Flow آنے والے WebSocket پیغامات کے اسٹریم کی نمائندگی کر سکتا ہے۔ ہر پیغام اسٹریم میں جاری کیا جاتا ہے، آپریٹرز کے ذریعے فلٹرنگ اور جمع سے گزرتا ہے، اور نتیجہ کلائنٹ کو بھیجا جاتا ہے۔ یہ نقطہ نظر Kotlin پروجیکٹس میں Reactor یا RxJava جیسی ری ایکٹیو لائبریریوں کی جگہ لے لیتا ہے۔

موجودہ RxJava کوڈ کے ساتھ Flow کی مطابقت kotlinx-coroutines-rx3 ماڈیول فراہم کرتا ہے۔ توسیعی فنکشن Flow.asObservable() Flow کو RxJava 3 کے Observable میں تبدیل کرتا ہے۔ الٹی تبدیلی — CompletableSource.asFlow()، Observable.asFlow()۔ یہ RxJava سے کوروٹینز میں منتقلی کو آسان بناتا ہے: پروجیکٹ کو مرحلہ وار دوبارہ لکھا جا سکتا ہے، کچھ پرتیں RxJava پر چھوڑ کر۔ تبدیلی کرتے وقت، cold/hot سیمنٹکس میں فرق پر غور کرنا ضروری ہے: Observable cold اور hot دونوں ہو سکتا ہے، Flow عام Flow کے لیے ہمیشہ cold اور SharedFlow کے لیے hot ہے۔

Flow کی خرابی کا انتظام اور جانچ

Flow میں خرابی کے انتظام کی ایک خاصیت ہے: اگر ٹرمینل آپریٹر سے پہلے flow بلڈر کے اندر استثنا ہوتا ہے، تو یہ catch میں منتقل ہو جاتا ہے۔ اگر بلڈر کے بعد کسی آپریٹر میں استثنا ہوتا ہے، تو اس آپریٹر کے بعد والا catch اسے پکڑتا ہے۔ retryWhen ایک شرط کے ساتھ سبسکرپشن کو دوبارہ آزمانے کی اجازت دیتا ہے: نیٹ ورک کی خرابی پر 3 بار تک دوبارہ کوشش کریں، لیکن CancellationException پر دوبارہ کوشش نہ کریں۔ Flow حالت پر منحصر خرابیوں کو ختم کرتا ہے کیونکہ یہ حالت محفوظ نہیں کرتا — یہ Observable کے مقابلے میں ڈیبگنگ کو آسان بناتا ہے، جہاں Subject اندرونی حالت محفوظ کرتا ہے۔

kotlinx-coroutines-test کے ساتھ Flow کی جانچ تاخیر کی نقل کرنے کے لیے TestDispatcher استعمال کرتی ہے۔ Turbine Flow کی جانچ کے لیے ایک مقبول کمیونٹی لائبریری ہے: test { } Flow شروع کرتا ہے، awaitItem() اگلی قدر کا انتظار کرتا ہے، awaitComplete() تکمیل کا انتظار کرتا ہے۔ Turbine ایک ڈیفالٹ ٹائم آؤٹ شامل کرتا ہے، جو جانچوں کو پھنسنے سے روکتا ہے۔ StateFlow کی جانچ کے لیے، تاریخی ترتیب میں قدر کی تصدیق کے ساتھ .testIn(scope) استعمال کریں۔