برنامه نویسیReactiveفقط یکAPIنیست. این یک الگوی کاملاً جدید برنامه نویسی است که مربوط بهstreamداده ها و گسترش تغییرات است.

RxJavaیک پیاده سازیReactiveبرای آوردن این مفهوم به سیستم عامل اندروید است. برنامه های اندرویدی مکان مناسبی برای شروع کاوش دردنیایواکنش پذیر هستند. RxAndroid یک کتابخانه که رویدادهای ناهمگامUIرا بسته بندی می کند تابهRxJavaبیشتر شبیه باشد.

در این آموزش برای برنامه نویسی واکنشی(Reactive) ، نحوه انجام موارد زیر را یاد خواهید گرفت:

• مفاهیم برنامه نویسی واکنش پذیر را درک میکنید.

• Observableرا تعریف میکنید.

• رویدادهای ناهمزمان مانندeventهای دکمه ای و تغییراتcontextمتن را به ساختارهایobservableتبدیل کنید.

• observable ها را تغییر شکل دهید و فیلتر کنید.

• استفاده ازthread هایRxدر اجرای کد.

• چندینobservableرا در یک جریان ترکیب کنید.

• تمامobservableهای خود را به سازه هایFlowableتبدیل کنید.

• ازRxJava برای افزودن ویژگی مورد علاقه به برنامه استفاده کنید.

خب امیدوارم که مفاهیم برایتان ساده باشد.برای مثال میخواهیم اپلیکیشن یافتن پنیر را پیاده سازی کنیم.

   سورس را از  اینجا  دانلود کنید

شروع کنیم

شما هم درCheeseActivity.ktو هم درCheeseAdapter.ktکار خواهید کرد. کلاسCheeseActivity از BaseSearchActivityرا مشتق میشود. کد هایBaseSearchActivityرا بخوانید و میبینید که ویژگی هایی برای شما پیاده سازی کرده است که در زیر آنها را بررسی میکنیم:

• showProgress(): تابعی برای نمایش نوار پیشرفت.
• hideProgress(): تابعی برای پنهان کردن نوار پیشرفت.
• showResult(result: List): متدی برای نمایش لیستی از پنیرها.
• cheeseSearchEngine: فیلدی است که نمونه ازCheeseSearchEngineرا برای ما میسازد. این یک متد جستجو است که وقتی می خواهید پنیرها را جستجو کنید آن را فراخوانی میکنید. این متد یک درخواست جستجوی متن را میگیرد و لیستی از پنیرها مطابق با جستجوی ما را برمی گرداند.

پروژه را روی دستگاهAndroidیا شبیه ساز خود اجرا کنید. شما باید یک صفحه جستجو با خالی را ببینید:

البته قرار نیست برای همیشه اینجوریبماند شما به زودی توابع واکنشی را به برنامه اضافه خواهید کرد. قبل از ایجاد اولین  observable  ابتدا کمی تئوری صحبت کنیم.

برنامه نویسیReactiveچیست؟

در برنامه نویسیimperative  یک عبارت یک بار ارزیابی می شود و مقدار به یک متغیر اختصاص می یابد:

var x = 2
var y = 3
var z = x * y // z is 6

x = 10
// z is still 6

از طرف دیگر ، برنامه نویسیreactiveهمه چیز برمیگردد به پاسخ تغییرات مقدار.

شما احتمالاً بعضی مواقع برنامه نویسیreactiveرا انجام داده اید  حتی اگر در آن زمان متوجه آن نبودید.

• تعریف  cell valueدرspreadsheetمشابه تعریف متغیرها در برنامه نویسیimperativeاست.
• تعریف عباراتcellدرspreadsheetمانند تعریف و کار باobservable هادر برنامه نویسیreactiveاست.

spreadsheetزیر را که مثال بالا را پیاده سازی می کند در نظر بگیرید:

spreadsheetبه سلولB1با مقدار 2 ، سلولB2با مقدار 3 و سلول سوم ،B3، با عبارتی اختصاص می دهد که مقدارB1را در مقدارB2ضرب می کند. هنگامی که مقدار هر یک از اجزای ارجاع شده در عبارت تغییر می کند ، تغییر مشاهده می شود و عبارت به صورت اتوماتیک درB3ارزیابی می شود:

به بیان ساده تر ، ایده برنامه نویسیreactiveوجود اجزایی است که تصویر بزرگتری را تشکیل می دهندکه می توان مشاهده کرد. و از برنامه خود بخواهید که هر زمان که اتفاق افتاد تغییرات را گوش کند و آنها را مصرف کند.

تفاوت بینRxJavaوRxKotlin

همانطور که احتمالاً می دانید ، به لطف سازگاری زبان کاتلینبا جاوا امکان استفاده از کتابخانه های جاوا در پروژه های کاتلین وجود دارد. اگر چنین است ، پس چراRxKotlinدر وهله اول ایجاد شده است؟RxKotlinیک پکیج کاتلین پیرامونRxJavaاست ، که همچنین بسیاری از توابع مفید برای برنامه نویسی  reactiveرا فراهم می کند.

در این مقاله ، ما به استفاده ازRxJavaخواهیم پرداخت زیرا درک مفاهیم اصلی این روش بسیار مهم است. با این حال ، هر آنچه یاد خواهید گرفت برایRxKotlinنیز صدق می کند.

توجه: به فایلbuild.gradleو به ویژه وابستگی های پروژه نگاهی بیندازید. به جز کتابخانه هایUI، این مجموعه شامل بسته هایRxKotlinوRxAndroidاست. نیازی نیست کهRxJavaرادر اینجا مشخص کنیم چون کهRxKotlinآن را دارد.

RxJava Observable Contract

RxJavaاز الگویObserverاستفاده می کند.

در الگویObserver، شما اشیایی دارید که دو رابط کلیدیRxJavaرا اجرا می کنند:ObservableوObserver. وقتی یک وضعیتObservableتغییر کند تمام اشیاObserverمشترک در آن مطلع می شوند.

از جمله متدهای موجود در اینترفیسObservable،subscribe()است كه یكObserverبرای شروع اشتراك تماس می گیرد.

از آن نقطه اینترفیس Observerدارای سه متد است که در صورت لزوم Observableآنها را فراخوانی می کند:

• onNext(T value)آیتم جدیدی از نوعTرا بهObserverارائه می دهد.
• onComplete()بهObserverاطلاع می دهد کهObservableارسال موارد را به پایان رسانده است.
• onError(Throwable e)بهObserverاطلاع می دهد کهObservableبا خطایی روبرو شده است

به عنوان یک قاعده  یک  Observableصفر یا چند مورد را منتشر می کند که می تواند تکمیل شود یا خطا داشته باشد.

به نظر پیچیده می رسد ، اما برخی از نمودارهایmarble ممکن است همه چیز را روشن کند

دایره نشان دهنده موردی است که ازobservableو بلوک سیاه نشان دهنده یک تکمیل یا خطا است. به عنوان مثال ، یک درخواست شبکه راobservableکنید. این درخواست معمولاً یک مورد (response) منتشر می کند و بلافاصله تکمیل می شود.

یک حرکت ماوس مختصات ماوس را منتشر می کند اما هرگز کامل نمی شود:

در اینجا می توانید چندین مورد را که منتشر شده اند مشاهده کنید اما هیچ بلوکی نشان نمی دهد ماوس خطایی را کامل یا ایجاد کرده است.

پس از اتمام یکobservable، هیچ مورد دیگری قابل انتشار نیست. در اینجا مثالی از رفتارهای بد مشاهده شده است که قراردادObservableرا نقض می کند:

این یکobservableبد است زیرا با انتشار یک مورد پس از علامت کامل شدن ، قراردادObservableرا نقض می کند.

نحوه ایجاد یکObservable

کتابخانه های بسیاری وجود دارد که به شما کمک می کند تقریباً از هر نوع رویدادیobservableبسازید. با این حال ، گاهی اوقات فقط باید یک مطالعه انجام بدید. علاوه بر این یک روش عالی برای یادگیری در مورد الگویObservableو برنامه نویسیreactiveاست!

با استفاده ازObservable.create()یکObservableایجاد خواهید کرد. در اینجا نمونه آن است:

Observable<T> create(ObservableOnSubscribe<T> source)

نمونه بالا خوب و مختصر است ، اما چه معنایی دارد؟ برای درک این نمونه ، باید بدانید کهObservableOnSubscribeچیست. اینinterfaceبا این قرارداد است:

public interface ObservableOnSubscribe<T> {
  void subscribe(ObservableEmitter<T> e) throws Exception;
}

بنابراین "منبعی" که شما برای ایجادObservableخود نیاز دارید ، بایدsubscribe() را در معرض دید قرار دهد ، که به نوبه خود برای تهیه یک فرستنده به عنوان یک پارامتر به هر آنچه که آن را فرا می خواند نیاز دارد. پس یک منتشر کننده چیست؟

اینترفیسRxJava Emitterمشابه اینترفیسObserverاست:

public interface Emitter<T> {
  void onNext(T value);
  void onError(Throwable error);
  void onComplete();
}

ObservableEmitterابزاری را برای لغو اشتراک فراهم می کند.

برای تجسم کل این وضعیت ، به یک شیر آب فکر کنید که جریان آب را تنظیم می کند. لوله های آب مانند یک  Observableاست  اگر وسیله ای برای ضربه زدن به آن دارید می تواند جریان آب را تحویل دهد. شما یک شیرآلات درست می کنید که می تواند خاموش و روشن شود ، مانند یکObservableEmitterاست و آن را به لوله های آب موجود درObservable.create()متصل می کنید. نتیجه یک شیر فانتزی خوب است. و البته ، شیر آب واکنش پذیر است ، زیرا به محض بستن آن ، جریان آب - داده ها - دیگر فعال نیستند.

یک مثال باعث می شود وضعیت انتزاعی نباشد و وضوح بیشتری داشته باشد. وقت آن است که اولینobservable های خود را ایجاد کنید!

کلیک های دکمه راObserveکنید

کد زیر را در کلاسCheeseActivityاضافه میکنیم

private fun createButtonClickObservable(): Observable<String> {
    // 2
    return Observable.create { emitter ->
        // 3
        searchButton.setOnClickListener {
            // 4
            emitter.onNext(queryEditText.text.toString())
        }

        // 5
        emitter.setCancellable {
            // 6
            searchButton.setOnClickListener(null)
        }
    }
}

importهای شما پس از وارد کردن کد فوق باید به صورت زیر باشد:

import io.reactivex.Observable
import kotlinx.android.synthetic.main.activity_cheeses.*

شما کلاسObservableصحیح را وارد کرده اید و ازKotlin Android Extensionsبرای دریافت منابع برای مشاهده اشیا استفاده می کنید.

توضیحات کد بالا:

1. شما تابعی را اعلام می کنید که Observableرا بازگرداند که رشته هایی را منتشر کند.
2. شما یکObservableرا باObservable.create()ایجاد کرده و با یکObservableOnSubscribeجدید آن راعرضه می کنید.
3. OnClickListenerرا درsearchButtonراه اندازی میکنید.
4. هنگامی که رویداد کلیک اتفاق می افتد ، از طریق فرستندهonNextرا فراخوانی میکنید و مقدار متن فعلیqueryEditTextرا به آن منتقل میکنید.
5. نگه داشتنreferenceها می تواند باعثmemory leakدر جاوا یا کاتلین شود. این عادت مفیدی است کهlistener هارا به محض اینکه دیگر نیازی به آنها نیست حذف کنید. اما وقتی در حال ایجادObservableخود هستید ، چه چیزی را صدا می کنید؟ به همین دلیل ،ObservableEmitterCancellable()را تنظیم کرده است.باoverrideکردنcancel()و هنگامی کهObservableاز بین می رود ، اجرای آن فراخوانی می شود ، مانند زمانی کهObservableکامل می شودیاunsubscribeمیشود.
6. برایOnClickListener، کد (null)setOnClickListener,listener  را حذف می کند

اکنون کهObservableخود را تعریف کردید ، بایدsubscriptionآن را تنظیم کنید. قبل از انجام این کار ، باید با یکinterfaceدیگر یعنیConsumerآشنا شوید. این یک روش ساده برای پذیرش مقادیری است که از یک فرستنده(emitter) وارد می شود.

public interface Consumer<T> {
  void accept(T t) throws Exception;
}

اینinterfaceزمانی مفید است که می خواهیدsubscriptionساده یکObservableرا تنظیم کنید.

Interfaceکه برایObservableمینویسیمبه چندین نسخه   subscribe()احتیاج داردکه پرارامترهایمختلفی دارند. به عنوان مثال ، اگر دوست دارید می توانید از یکObserverکامل عبور کنید ، اما لازم است تمام متد های لازم را پیاده سازی کنید.

اگر تمام آنچه در خارج ازsubscription  نیاز دارید این است کهobserverبه مقادیر ارسال شده بهonNext()پاسخ دهد, می توانید از نسخهsubscribe()استفاده کنید که یک single Consumerرا در بر می گیرد.

این کار را دقیقاً هنگامsubscribeدرactivityخود در ()onStartانجام خواهید داد.

کد زیر را بهCheeseActivity.ktاضافه میکنیم:

override fun onStart() {
    super.onStart()
    // 1
    val searchTextObservable = createButtonClickObservable()

    searchTextObservable
            // 2
            .subscribe { query ->
                // 3
                cheeseSearchEngine.search(query)?.let { showResult(it) }
            }
}

توضیحی درباره مراحل میدهیم:

1. ابتدا با فراخوانی متدی که به تازگی نوشتید ، یک  observableایجاد کنید.
2. مشترک شدن در  observable  باsubscribe() و ارائه یکConsumerساده.
3. در آخر  جستجو را انجام دهید و نتایج را نشان دهید.

برنامه را اجرا کنید. چند حرف وارد کنید و روی دکمه جستجو ضربه بزنید. پس از یک تاخیر شبیه سازی شده (بهCheeseSearchEngineمراجعه کنید) ، باید لیستی از پنیرها را که مطابق با درخواست شما هستند ، ببینید.

RxJava Threading Model

شما اولین تجربه برنامه نویسیreactiveرا داشته اید. یک مشکل وجود دارد: با لمس دکمه جستجو ، رابط کاربر برای چند ثانیه مسدود می شود.

درAndroid Monitorممکن است متوجه خط زیر شوید:

> 08-24 14:36:34.554 3500-3500/com.raywenderlich.cheesefinder I/Choreographer: Skipped 119 frames!
  The application may be doing too much work on its main thread.

این اتفاق می افتد زیرا جستجو بر رویmain threadاجرا می شود.اگر جستجو برای انجام درخواست شبکه انجام شود ،Androidبا یک استثنا(crash)NetworkOnMainThreadExceptionبرنامه را میبندد. وقت آن است که آن را برطرف کنیم.

یکی از توضیحاتRxJavaاین است که به صورت پیش فرض  multi-threadاست شبیهAsyncTask. با این حال ، اگر روش دیگری مشخص نشده باشد ،RxJavaتمام کارها را در همانthreadانجام می دهد که از آن فراخوانی شده است.

می توانید این عمل را با عملگرهایsubscribeOnوObserveOnتغییر دهید.

subscribeOnقرار است فقط یک بار در زنجیره اپراتورها فراخوانی شود. اگر اینگونه نباشد ، اولین فراخوانی اجرا می شود. subscribeOnیکthreadرا مشخص می کند کهobservableدر آنsubscribeمی شود (یعنی ایجاد می شود). اگر از یکobservableاستفاده می کنید کهeventها را از نمایAndroidمنتشر می کنند ، باید مطمئن شوید کهsubscriptionدرthread Android UIانجام شده است. از طرف دیگر اشکالی ندارد هر تعداد بار که خواستید در زنجیرهObserveonرا فراخوانی کنید.

ObservOnیکthreadرا مشخص می کند که اپراتورهای بعدی در این زنجیره اجرا می شوند.  مثال زیر را ببینید:

myObservable // observable will be subscribed on i/o thread

      .subscribeOn(Schedulers.io())

      .observeOn(AndroidSchedulers.mainThread())

      .map { /* this will be called on main thread... */ }

      .doOnNext{ /* ...and everything below until next observeOn */ }

      .observeOn(Schedulers.io())

      .subscribe { /* this will be called on i/o thread */ }

مفیدترینschedulerها عبارتند از:

• ()Schedulers.io: مناسب برای کارهای ورودی و خروجی مانند درخواست شبکه یا عملکرد دیسک.
• Schedulers.computation(): با وظایف محاسباتی مانند حلقه های رویداد و پردازش callbackها بهترین عملکرد را دارد
• AndroidSchedulers.mainThread() :عملگرهای بعدی را درUI threadاجرا می کند.
•  

اپراتورMap

اپراتورmapتابعی را برای هر مورد منتشر شده توسط یک observableاعمال می کند و یکobservableدیگر را که نتایج حاصل از آن فراخوانی های تابع را منتشر می کند  برمی گرداند. برای رفع مشکلthreadنیز به این مورد احتیاج دارید.

اگرobservableبه نامnumbersدارید که موارد زیر را منتشر می کند:

و اگرmapرا به صورت زیر اعمال می کنید:

numbers.map { number -> number * number }

نتیجه به شرح زیر خواهد بود:

این یک روش مفید برای تکرار چندین مورد با کد کم است. بگذارید استفاده کنیم!

متد ()onStartکلاسCheeseActivityبه صورت زیر تغییر میدهیم:

override fun onStart() {
    super.onStart()

    val searchTextObservable = createButtonClickObservable()

    searchTextObservable
            // 1
            .subscribeOn(AndroidSchedulers.mainThread())
            // 2
            .observeOn(Schedulers.io())
            // 3
            .map { cheeseSearchEngine.search(it)!! }
            // 4
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe {
                showResult(it)
            }
}

کد بالا را مرور میکنیم:

1. ابتدا مشخص میکنیم که کد به جایI/O threadباید رویmain threadشروع شود. درAndroid، همه کدهایی که باViewکار می کنند باید رویmain threadاجرا شوند.
2. مشخص میکنیم که عملگر بعدی باید رویI/O threadفراخوانی شود.
3. برای هرsearch query، لیستی از نتایج را برمی گردانید.
4. در آخر ، اطمینان حاصل میکنیم که نتایج به لیستmain threadمنتقل شده اند.

پروژه را اجرا کنید. اکنونUIباید پاسخگو باشد حتی زمانی که جستجو در حال انجام است.

نمایشProgressBarباdoOnNext

برای این منظور شما به یک عملگرdoOnNextنیاز دارید. doOnNextیکConsumerمی گیرد و به شما امکان می دهد هر بار که موردی توسطobservableمنتشر می شود کاری انجام دهید.

در کلاسCheeseActivity , ()onStartرا مانند زیر تغییر میدهیم:

override fun onStart() {
    super.onStart()

    val searchTextObservable = createButtonClickObservable()

    searchTextObservable
            // 1
            .observeOn(AndroidSchedulers.mainThread())
            // 2
            .doOnNext { showProgress() }
            .observeOn(Schedulers.io())
            .map { cheeseSearchEngine.search(it)!! }
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe {
                // 3
                hideProgress()
                showResult(it)
            }
}

توضیحات کد بالا:

1. اطمینان حاصل میکنیم که اپراتور بعدی در زنجیره بر رویmain threadاجرا می شود.
2. عملگرdoOnNextرا اضافه میکنیم تا هر زمان که مورد جدیدی منتشر شد ()showProgressفراخوانی شود.
3. فراموش نکنید هنگامی که می خواهید نتیجه ای را نمایش دهید ()hideProgressرا فراخوانی کنید.

پروژه اجرا کنید. هنگام شروع جستجو باید ببینیدprogress barظاهر می شود:

Observe Text Changes

اگر بخواهید وقتی کاربر برخی از متن ها را درست مثلGoogleتایپ می کند ، جستجو را انجام دهید چه؟

ابتدا ، بایدsubscribeکنیم به textchangesازTextView. متد زیر را به کلاسCheeseActivityاضافه میکنیم:

// 1
private fun createTextChangeObservable(): Observable<String> {
    // 2
    val textChangeObservable = Observable.create<String> { emitter ->
        // 3
        val textWatcher = object : TextWatcher {

            override fun afterTextChanged(s: Editable?) = Unit

            override fun beforeTextChanged(s: CharSequence?, start: Int, count: Int, after: Int) = Unit

            // 4
            override fun onTextChanged(s: CharSequence?, start: Int, count: Int, after: Int) {
                s?.toString()?.let { emitter.onNext(it) }
            }
        }

        // 5
        queryEditText.addTextChangedListener(textWatcher)

        // 6
        emitter.setCancellable {
            queryEditText.removeTextChangedListener(textWatcher)
        }
    }

    // 7
    return textChangeObservable
}

کد بالا را خط به خط بریم جلو:

1. متدی را از نوعobservableتعریف میکنیم برای تغییرات متن.
2. تعریف متغیرtextChangeObservableبا استفاده از  ()createکه نوع آنObservableOnSubscribeاست.
3. هنگامی که یکobserverاز نوع اشتراکی تعریف میکنیم  اولین کاری که باید انجام شود ایجاد یکTextWatcherاست.
4. ما با قسمت ()beforeTextChangedوafterTextChanged()کاری نداریم اما در قسمتonTextChanged()مقدار متن جدید را به یکobserverمنتقل میکنیم.
5. با فراخوانی ()addTextChangedListener,watcherرا بهTextViewخود اضافه کنید.
6. فراموش نکنید کهwatcherخود را حذف کنید. برای این کار ، ازemitter.setCancellable()و نوشتن مجددcancel()  برای فراخوانی    ()removeTextChangedListenerاستفاده میکنیم.
7. سرانجامobservableایجاد شده را برگردانید.

متغیرsearchTextObservableکه در متدonStart(),CheeseActivityتعریف کردیم را به صورت زیر تغییر دهید:

val searchTextObservable = createTextChangeObservable()

برنامه را اجرا کنید. وقتی شروع به تایپ متن درTextViewمی کنید باید جستجو شروع شود:

درخواستها را براساس طول فیلتر میکنیم(Filter Queries by Length)

منطقی نیست که با وارد کردن یک حرف شروع کنیم به جستجو کردن. برای رفع این مشکل ، از اپراتور قدرتمندfilterاستفاده میکنیم. filterفقط از مواردی عبور می کند که شرایط خاصی را برآورده می کنند. فیلتر یکpredictرا می گیرد در این حالتPredicateیک رشته را می گیرد و اگر طول رشته دو یا چند کاراکتر باشد ،trueبرمی گردد.

return textChangeObservableدر متدcreateTextChangeObservable()را با کد زیر جایگزین میکنیم:

return textChangeObservable.filter { it.length >= 2 }

همه کارها دقیقاً مشابه قبل خواهند بود ، با این تفاوت که نمایش داده های متنی با طول کمتر از 2 به زنجیره ارسال نمی شوند.

برنامه را اجرا کنید ؛ شما باید ببینید که فقط در هنگام تایپ کردن کاراکتر دوم جستجو شروع می شود:

Debounce Operator

ما نمی خواهیم هر بار که درخواست یک کاراکتر تغییر کرد درخواست جدیدی به سرور ارسال کنیم.

debounceیکی از آن اپراتورهایی است که قدرت واقعی الگویreactiveرا نشان می دهد. دقیقاً مانند عملگر فیلترdebounce مواردی را که توسط  observableمنتشر می شود فیلتر می کند. debounceبعد از انتشار هر مورد برای مورد بعدی مدت زمان مشخصی منتظر می ماند. اگر طی این انتظار هیچ موردی منتشر نشود ، سرانجام آخرین مورد منتشر می شود:

در ()createTextChangeObservable، اپراتورdebounceرا درست در زیر فیلتر اضافه میکنیم تا دستور بازگشت مانند کد زیر باشد:

return textChangeObservable
        .filter { it.length >= 2 }
        .debounce(1000, TimeUnit.MILLISECONDS) // add this line

debounce قبل از انتشار آخرین متن 1000 میلی ثانیه منتظر می ماند.

اپراتورMerge

شما یکobservableکه نسبت به کلیک دکمه ها واکنش نشان می دهد شروع کردید و سپس یکobservableرا که به تغییرات فیلد متن واکنش نشان می دهد پیاده سازی کردید. اماobservableشما در برابر هر دو واکنش نشان می دهد؟

اپراتورهای زیادی برای ترکیب  observableها وجود دارد. ساده ترین و مفیدترین آن هاmergeاست.

mergeموارد را از دو یا چندobservableمی گیرد و آنها را در یکobservableقرار می دهد:

متدonStart()را به شکل زیر تغییر میدهیم:

override fun onStart() {
    super.onStart()

    val buttonClickStream = createButtonClickObservable()
    val textChangeStream = createTextChangeObservable()

    val searchTextObservable = Observable.merge<String>(buttonClickStream, textChangeStream)


    searchTextObservable
            // 1
            .observeOn(AndroidSchedulers.mainThread())
            // 2
            .doOnNext { showProgress() }
            .observeOn(Schedulers.io())
            .map { cheeseSearchEngine.search(it)!! }
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe {
                // 3
                hideProgress()
                showResult(it)
            }
}

برنامه خود را اجرا کنید. با قسمت متن و دکمه جستجو بازی کنید. جستجو وقتی شروع می شود که تایپ کردن دو یا چند نماد را تمام کنید یا وقتی دکمه جستجو را فشار می دهید.

اپراتورFlowable

با انتشارRxJava2، چارچوب کاملاً از ابتدا برای حل برخی از مشکلات که در کتابخانه اصلی به آنها پرداخته نشده بود دوباره طراحی شده است. یکی از موضوعات بسیار مهمی که در بروزرسانی به آن پرداخته شده ، ایدهbackpressureاست.

Backpressureاین مفهوم است که یکobservableسریعتر از آنچهconsumerاز عهده آن برمی آید داده منتشر می کند. مثالFirehoseتوییتر را در نظر بگیرید که با افزودن توییتر به پلتفرم توییتر  مرتباً توییت هایی را منتشر می کند. اگر بخواهید از  observableاستفاده کنید  تا زمانی که حافظه دیگری در دسترس نباشد برنامه شما خراب می شود و با استفاده ازAPI firehoseامکان استفاده از آنها وجود ندارد. Flowableها این مورد را در نظر می گیرند و به شما اجازه می دهند یک استراتژیBackPressureمشخص کنید تا بهFlowableبگویید که چگونه می خواهید Consumerاز موارد منتشر شده سریعتر از مصرف استفاده کند.

استراتژیBackPressure:

• BUFFERموارد را به همان روشRxJava 1هندل می کند اما می توانید اندازه بافر را نیز اضافه کنید.
• DROPهر موردی راکهconsumerقادر به اداره آن نیست را رها می کند.
• ERRORوقتی پایین دست نمی تواند ادامه دهد ، خطایی ایجاد می کند.
• LATESTفقط آخرین مورد منتشر شده توسطonNextرا جایگزین مقدار قبلی می کند.
• MISSINGبدون بافر کردن یا رها کردن در رویداد های بعدی.

تبدیل  ObservableبهFlowable

زمان آن است که با استفاده از این دانش جدید از استراتژی  backpressure،observableهای فوق را به  flowableتبدیل کنیم. ابتدا observableهایی را که به برنامه خود اضافه کرده اید در نظر بگیرید. شما یک  observableدارید که با کلیک یک دکمه موارد را منتشر می کند و دیگری از ورودی صفحه کلید. با در نظر گرفتن این دو مورد در حالت اول تصور کنید که می توانید ازLATESTو در حالت دوم می توانید ازBUFFERاستفاده کنید.

CheeseActivity.ktرا باز کرده وobservableهای خود را به مانند زیر تغییر میدهیم:

val buttonClickStream = createButtonClickObservable()
        .toFlowable(BackpressureStrategy.LATEST) // 1

val textChangeStream = createTextChangeObservable()
        .toFlowable(BackpressureStrategy.BUFFER) // 2

1. با استفاده ازLATEST BackpressureStrategyکلیک دکمه را به یکflowableتبدیل کردیم.
2. با استفاده ازBUFFER BackpressureStrategyجریان تغییر متن ورودی را یکFlowableتبدیل کردیم.

در آخر ، عملگر  mergeرا تغییر میدهیم تا ازFlowableنیز استفاده کند:

val searchTextFlowable  = Flowable.merge<String>(buttonClickStream,
 textChangeStream)

اکنون برای استفاده از مقدار جدیدsearchTextFlowable به جای مقدار قبلیObservableجایی که آن را فراخوانی کرده ایم نامش را تغییر میدهیم.

searchTextFlowable
        // 1
        .observeOn(AndroidSchedulers.mainThread())
        // 2
        .doOnNext { showProgress() }
        .observeOn(Schedulers.io())
        .map { cheeseSearchEngine.search(it)!! }
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe {
            // 3
            hideProgress()
            showResult(it)
        }

برنامه را دوباره اجرا کنید و باید یک برنامه در حال کار را ببینید که هیچ یک از مشکلات observableرا ندارد.

Maybe

Maybeیک محاسبه است که یا یکvalue singleبدون مقدار یا خطا منتشر می کند. آنها برای مواردی مانند به روزرسانی و حذف پایگاه داده مناسب هستند. در اینجا شما یک ویژگی جدید را با استفاده ازMaybeبه نوع خاصی از پنیر از برنامه اضافه می کنید و ازMaybeبرای انتشار مقدار هیچ استفاده می کنید.

به کلاسCheeseAdapterمیرویم و کد زیر را درonBindViewاضافه میکنیم:

// 1
Maybe.create<Boolean> { emitter ->
    emitter.setCancellable {
        holder.itemView.imageFavorite.setOnClickListener(null)
    }

    holder.itemView.imageFavorite.setOnClickListener {
        emitter.onSuccess((it as CheckableImageView).isChecked) // 2
    }
}.toFlowable().onBackpressureLatest() // 3
        .observeOn(Schedulers.io())
        .map { isChecked ->
            cheese.favorite = if (!isChecked) 1 else 0
            val database = CheeseDatabase.getInstance(holder.itemView.context).cheeseDao()
            database.favoriteCheese(cheese) // 4
            cheese.favorite // 5
        }
        .subscribeOn(AndroidSchedulers.mainThread())
        .subscribe {
            holder.itemView.imageFavorite.isChecked = it == 1 // 6
        }

توضیحات کد بالا:

1. یکactionازMaybeمیسازیم.
2.  چک کردن وضعیت در زمان.success
3. تبدیلMaybeبهFlowable
4. انجام به روزرسانی را روی جدول پنیرها.
5. نتیجه عملیات را برمیگرداند.
6. از نتیجه حاصل از انتشار استفاده میکنیم تا تصویر قلب خالی به قلب پر شده تغییر داده شود.

توجه: بهتر است که ازMaybeبرای عملیات حذف استفاده کنید اما به عنوان مثال در اینجا می توانید یک پنیر راfavorateکنید.

RxJava2 & Null

مقدارNullدیگر درRxJava2پشتیبانی نمی شود.استفاده ازnullباعث ایجادNullPointerExceptionمیشود.

RxJava and Activity/Fragment lifecycle

متدهایsetCancellableقابل تنظیم را به خاطر می آورید؟ تا زمانی کهobservableلغو نشود  آنهاunsubscribeنخواهند شد.

فراخوانیObservable.subscribe()یکDisposableرا برمیگرداند. Disposableیکinterfaceاست که دارای دو متد است:

public interface Disposable {
  void dispose();  // ends a subscription
  boolean isDisposed(); // returns true if resource is disposed (unsubscribed)
}

propertyزیر را بهCheeseActivityاضافه کنید:

private lateinit var disposable: Disposable

درonStart()مقدار برگشتیsubscribe()را با استفاده از کد زیر در disposableتنظیم کنید (فقط خط اول تغییر می کند):

disposable = searchTextFlowable // change this line
        // 1
        .observeOn(AndroidSchedulers.mainThread())
        // 2
        .doOnNext { showProgress() }
        .observeOn(Schedulers.io())
        .map { cheeseSearchEngine.search(it)!! }
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe {
            // 3
            hideProgress()
            showResult(it)
        }

از آنجا کهobservableدر ()onStart,subscribeشد ،onStop()مکانی عالی برایunsubscribeخواهد بود.

کد زیر را بهCheeseActivity.ktاضافه کنید:

override fun onStop() {
    super.onStop()
    if (!disposable.isDisposed) {
        disposable.dispose()
    }
}

برنامه را اجرا کنید. خودتان هیچ تغییری را مشاهده نخواهید کرد ، اما اکنون این برنامه با موفقیت از RxJava memory leakجلوگیری می کند.

سوالات و مشکلات خود را در قسمت کامنت ها مطرح کنید تا ما به آنها پاسخ دهیم.