LiveData

just

MutableLiveData

fun resetService(service: Service, usage: Usage): Single<Foo> =
        getSubscriber()
            .map { it.subscriber }
            .flatMap { subscriber ->
                Single
                    .fromCallable {
                        val credit = usage.credit ?: 0.0
                        if (subscriber.type == Type.CREDIT && creditAmount < service.resetPrice) {
                            throw InsufficientCreditException(creditAmount, service.resetPrice)
                        }

                        val product = getProduct(cache, service.productId)
                        val usageManagement = product?.usageManagement ?: error("UsageManagent is null")
                        val resetProductId = product.resetProductIdOverride ?: usageManagement.resetProductId

                        tupleOf(usageManagement.action, resetProductId)
                    }
                    .doOnSuccess {
                        store.setProcessingSubscriber(subscriber.id, isProcessing = true)
                    }
                    .flatMap { (action, resetProductId) ->
                        setService(action, resetProductId, null, null, null, null)
                    }
                    .doOnErrorOrCancel {
                        store.setProcessingSubscriber(subscriber.id, isProcessing = false)
                    }
            }
            .doOnSuccess { updateManager.enqueueUpdate() }

onErrorReturn

override fun loadDrillTypes(): Observable<ArrayList<DrillsType>> {

        val list = ArrayList<DrillsType>()

        firebaseFirestore.collection("drilltypes")
            .get()
            .addOnSuccessListener { documents ->
                for (document in documents) {
                    val doc = document.toObject(DrillsType::class.java)
                    list.add(doc)
                }
            }
            .addOnFailureListener { exception ->
                Log.w("TAG", "Error getting documents: ", exception)
            }
        return Observable.just(list)   // delay(2, TimeUnit.SECONDS)
    }

compositeDisposable += repository.loadDrillTypes()
            .subscribeWith(object : DisposableObserver<ArrayList<DrillsType>>() {

            override fun onError(e: Throwable) {
                //if some error happens in our data layer our app will not crash, we will
                // get error here
            }

            override fun onNext(data: ArrayList<DrillsType>) {
                Log.d("TAG", data.toString())
            }

            override fun onComplete() {
                Log.d("TAG", "COMPLETE")
            }
        })
    }

fun connect(bluetoothDevice: BluetoothDevice) : Observable<RxBleConnection>

Observable

Single

Observable

suspend fun connect(bluetoothDevice : BluetoothDevice) : RxBleConnection

connectObservable.awaitFirst()

share

buffer

private val dealerAction: Single<DealerAction> = carDealersRepo.selectedCarDealer()
    .map {
        it.toDealerAction()
    }

val dealerButtonTitle: LiveData<String> = dealerAction
    .map { it.toButtonTitle() }
    .toObservable()
    .toLiveData()

selectedCarDealer

private val dealerButtonTrigger: PublishSubject<Unit> = PublishSubject.create()

init {
    dealerButtonTrigger.withLatestFrom(dealerAction.toObservable())
        .subscribe { (_, action) ->
            action.handle()
        }
}

dealerButtonTrigger.onNext(Unit)

flatMap

withLatestFrom

withLatestFrom

doOnSubscribe

doOnSubscribe

val observer = object : DisposableObserver<DrillsType>() {

            override fun onError(e: Throwable) {
                Log.d("TAG", "error " + e.message)
            }

            override fun onNext(data: DrillsType) {
                //onnext
            }

            override fun onComplete() {
                Log.d("TAG", "COMPLETE")
            }
        }
        remoteDataSource.loadDrillTypes().subscribeWith(observer)
    }

Can't error out

Single

BehaviorSubject

Observable

just(1)
    .flatMap { just(2).delay(1, SECONDS).startWith(it) }
    .subscribe { println(it) }

CompositeDisposable

.subscribe

io

computation

.subscribe()

onError

.subscribe({ }, { })

Exception in thread "UI thread @coroutine#1" java.lang.ClassCastException: io.reactivex.Observer$Subclass1 cannot be cast to com.nikolam.basketpro.ui.drills.selection.DrillsSelectionViewModel$fetchDrillTypes$1$disposable$1
	at com.nikolam.basketpro.ui.drills.selection.DrillsSelectionViewModel$fetchDrillTypes$1.invokeSuspend(DrillsSelectionViewModel.kt:37)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:56)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

BehaviorRelay.value