@Test
   fun `assert observable fires only once ever`() {
       val testObservable = Observable.just(true)

       testObservable.test().assertValue(true)

       testObservable.test().assertNoValues()
   }

val isRead = AtomicBoolean(false)
val testObservable = Maybe.fromCallable { true.takeUnless { isRead.getAndSet(true) } }.toObservable()

onNext

subject

subject

subscribe

subject

ObservableEmitter.setDisposable(Disposables.fromAction {})

Observable.doOnDispose{}

Single.fromCallable {
    someBlockingCode()
}

onComplete()

onError()

LiveDataReactiveStreams

Boolean

Observable<Boolean>

PublishSubject

viewModel.isSomethingTrue

onError()

val testNulls = Observable.create<String> {
        throw NullPointerException()//case 1
//        it.onNext(null!!) // case 2
        it.onNext("Ahmed")
    }

upload

.flatMap(::someFunction, false, 5)

requestData()

Observable.just(Unit)
                .delay(REPEAT_DELAY, TimeUnit.MILLISECONDS)
                .repeat()
                .async(job)
                .subscribe({ requestData() },
                        {
                            Timber.e(it, "subscribeError")
                        }
                )

Observable

doOnUnsubscribe

Observable<Integer> ob = Observable.timer(30,TimeUnit.SECONDS).just(1);

        ob.subscribe(n-> System.out.println(n));

PublishSubject

Flow

BroadcastChannel

Flowable

@Test fun `test poll`() {
    open class MyClient {
        fun fetchItemFromServer(): Single<Boolean> = Single.just(true)
    }

    val testScheduler = TestScheduler()
    val client = mock<MyClient>()

    whenever(client.fetchItemFromServer())
            .thenReturn(Single.just(false), Single.just(true))

    val test = client.fetchItemFromServer()
            .repeatWhen { it.delay(10, TimeUnit.SECONDS, testScheduler) }
            .distinctUntilChanged()
            .test()

    testScheduler.advanceTimeBy(10, TimeUnit.SECONDS)

    test.assertResult(false, true)
}

Maybe.just(list)
    .flatMapSingle { repository.getStuffList(id) }
    .map { stuffList -> stuffList.filter { stuff -> list.contains(stuff) } }
    .flatMapCompletable { stuffAfterFilter ->
         secondRepository.doThing(stuffAfterFilter)
              .doOnError { Completable.error(IllegalStateException("failure")) }
              .andThen { repository.deleteStuff(stuffAfterFilter) }
     }
     .andThen(repository.getStuffList(id))

repository.getStuffList(id)

stuffList2

stuffList1

whenever(repository.getStuffList("id")).thenReturn(Single.just(stuffList1))
whenever(secondRepository.doThing(any())).thenReturn(Completable.complete())
whenever(repository.deleteStuff(any())).thenReturn(Completable.complete())
whenever(repository.getStuffList("id")).thenReturn(Single.just(stuffList2))

fun getUserInfo() : Single<UserInfo>

fun save(token: Token) : Completable

fun initialize() {
    getUserInfo()
        .flatMap { 
            // Get token from UserInfo : val token = userInfor.getToken()
            // Call save(token: Token) method to save data
        }
        .subscribeOn(<http://Schedulers.io|Schedulers.io>())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribeBy(
            onSuccess = {
                // HERE:  Using UserInfo to do somtehing
            },
            onError = {
                Log.d(it)
            }
        )
}

flatMap

toObservable()

Completable.toObservable()

Returns an Observable which when subscribed to subscribes to this Completable and relays the terminal events to the subscriber.

Completable.complete()
    .toObservable<Any>()
    .subscribe { }

Result

Disposable

Single