class ChatViewModel(private val messageManager: MessageManager): ViewModel {
  private val binder: PublishProcessor<MessageEvent> = PublishProcessor.create()
  private val state =
    binder
      .flatMap{ messageManager.editMessage(it.messageId, it.newText) }
      // #1
      .replay(1)
      .autoConnect(0)

  fun editMessageClick(messageId: Long, newText: String) {
    binder.onNext(MessageEvent(messageId, newText))
  }
  override fun state(): Observable = state
}

class MessageManager {
  fun editMessage(messageId, newText) = 
    database.update(messageId, newText, State.SENDING)
      .flatMap { apiClient.editMessage(...)}
      .flatMap { database.update(messageId, newText, State.SENT )}
      .asEvents()
}

data class MessageEvent(val messageId: Long, val newText: String)

A.skipUntil(B.ignoreElements().concatWith(just(Unit)))

A

B

return Observable.just(1)
    .doOnSubscribe{
        this.disposable = observable.just(2)
            .subscribe({...}, {...})
    }

BiFunction

val one = Observable.fromIterable(listOf(1,2,3))
        val two = Observable.fromIterable(listOf(4,5,6))
        Observable.combineLatest(one, two, BiFunction{ p1:Int, p2:Int -> p1+p2})

fun printCurrentThread(tag: String) = println("$tag: ${Thread.currentThread().name.substringBefore("-")}")

fun main(args: Array<String>) {
	printCurrentThread("main thread name")

	val subject = BehaviorSubject.create(Unit)

	subject
			.subscribeOn(Schedulers.computation())
			.subscribe {
				printCurrentThread("subscribe onNext")
			}

	Thread.sleep(100L)

	subject.onNext(Unit)

	Thread.sleep(100L)
}

// The author said the code gives
   main thread name: main
   subscribe onNext: RxComputationScheduler
   subscribe onNext: main

The author explains:
Correct answer is A because the default value is based on the subscribeOn (computation thread) and .onNext() of the subject is called from the main thread. Therefore the second emission will ignore the subscribeOn, change thread and will be emitted at the main thread.

main thread name: main
subscribe onNext: main

Where the first RxComputationScheduler onNext()

fun parseObservable(f: File): Observable<Item> { ... }

Observable.fromIterable(List<File>)
                .concatMapEagerDelayError({ parseObservable(it) }, true)
                .toList()
                .subscribeOn(<http://Schedulers.io|Schedulers.io>())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe { items -> ... }

actionIntent()

loadFile()

Publisher<T>

buildPublisher().subscribe(subscriber)

buildPublisher()

Flowable.create<T>({ emitter -> 
  try {
    buildPublisher().subscribe(
        object : Subscriber<T> {
            override fun onComplete() {emitter.onComplete() }
            override fun onSubscribe(s: Subscription) { s.request(Long.MAX_VALUE) }
            override fun onNext(t: T) { emitter.onNext(t) }
            override fun onError(t: Throwable) { emitter.onError(t) }
        }
  } catch (e: Exception) { emitter.onError(e) }
}, BackpressureStrategy.BUFFER)
    .subscribeOn(<http://Schedulers.io|Schedulers.io>())
    .subscribe(subscriber)

Flowable

flatMapSingle

Flowable

val data = MutableLiveData<RegisterResponse>()
 disposables.add(
            ServiceRepo.register(firstName, lastName, mobileno, email, password)
                .subscribeOn(<http://Schedulers.io|Schedulers.io>())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe({
                    data.value = it
                }, { throwable ->
        val error: HttpException = throwable as HttpException
                    if (error.code() == 400) {
                        val jObj = error.response().errorBody() as RegisterResponse//the issue here, i can't cast it 
                        data.value = jObj
                    }

                })
        )

val a: Completable
val b: Completable
val c: Completable

val merged = Completable.merge(listOf(a,b))

merged

c

merged

c

c

fun doThings(): Maybe<Boolean> {
    Log.d("TEST", "HIT")
    return Maybe.just(true) 
}

return Maybe.just("")
    .map { false }
    .switchIfEmpty(doThings())

disposables.add(
            ServiceRepo.getProductReviews(context, pruductId)
                .subscribeOn(<http://Schedulers.io|Schedulers.io>())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe({
                  //how i can get them all here ♥
                }, { throwable ->
                    Log.e("tag", throwable.message)
                })
        )

Single.fromCallable {
 getUser().isVerified
}
.flatMapObservable { 
 if(it) Observable.just(it)
else getUserObservable().map { it.isVerified }
}
.subscribe ({
   if(it) view.showVerified
})

.doOnError

@Test
    fun `validate test`() {
        val scheduler = TestScheduler()

        val upstream = Flowable.fromCallable { "foo" }
            .subscribeOn(scheduler)
            .observeOn(scheduler)
            .delay(2, TimeUnit.SECONDS)

        val implementation = upstream
            .map { true }
            .startWith(false)
            .subscribeOn(scheduler)
            .observeOn(scheduler)
            .doOnEach { println(it) }
            .test()

        implementation
            .assertNotTerminated()
            .assertNoValues()

        scheduler.advanceTimeBy(1, TimeUnit.SECONDS)

        implementation
            .assertNotTerminated()
            .assertValues(false)

        scheduler.advanceTimeBy(2, TimeUnit.SECONDS)

        implementation
            .assertNotComplete()
            .assertValues(false, true) /* assertion fails*/

        /*
        * Output:
        * 
        * OnNextNotification[false]
        * 
        * java.lang.AssertionError: Value count differs; expected: 2 [false, true] but was: 1 [false] (latch = 1, values = 1, errors = 0, completions = 0)
        *   Expected :2 [false, true] 
        *   Actual   :1 [false] (latch = 1, values = 1, errors = 0, completions = 0)
        */
    }

blockingGet()