val events = Observable.just("event1", "event2", "event3")
  val initialState = Single.just(emptyList<String>())
  // given a current state produces next state's Single
  val reducerFactory = { currentState: List<String>, event: String ->
    Single.fromCallable { /* do work */ currentState.plus(event) }
  }
  events
    .scan(
      initialState,
      { currentStateSingle, event ->
        currentStateSingle
          .flatMap { currentState -> reducerFactory( currentState, event) }
          // required to avoid resubscription on all previously emitted single's on each new scan iteration
          .cache() // (1)
      }
    )
    .flatMapSingle { it }
    .subscribe { state -> println("state updated to $state") }

(1)

Singles

state

Completable

AppAuth

performTokenRequest

doOnSuccess

doNext

Completable

private fun login(
authorisationResponse: AuthorisationResponse
): Completable {
  return Completable.defer {
	// do some validations..
	if (authorisationResponse == null) {
	    return@defer Completable.error(...)
	}

	// app auth stuff
	val tokenRequest = authorizationResponse.createTokenExchangeRequest()
	val clientAuthentication ....

	// performing token request, AppAuth library works with callbacks, so I'm
	// curious how to "pause" my Rx flow until I get result from that callback
	// and then I need to continue completable

	return@defer ??? 
	  authorisationService.performTokenRequest(tokenRequest,clientAuthentication)
	{ response, ex -> 
		// callbacks
		if (response != null) {
			//... I'm going to save token here
            authStateManager.saveToken(response)
		}

	.doOnSuccess or .doNext (continueLoginFlow)
  }
}

private fun continueLoginFlow(): Completable {
    return authStateManager.getToken()
           .map { ... }
           .flatMap { ... }
           .doOnSuccess { ... } 
           ...

	...
}

val subject = PublishSubject.create<Int>()
    Observable.merge(
      Observable.just(1,2,3).doOnSubscribe { println("subscribed to observable") },
      subject.doOnSubscribe { println("subscribed to subject") }
    ).subscribe {
      println("received $it")
    }

subscribed to observable
received 1
received 2
received 3
subscribed to subject

subscribed to observable
subscribed to subject
received 1, 2, 3

Observable.just(1,3,4).doOnLast { subject.onNext(88) }

subscriberOn(Schedulers.single())

publishSubject.concatMap { Observable.delay(3000) }

Flow.buffer

onBackpressureBuffer

Observable

Single

Observable.zip(repo1.getStream(), repoSingle.getStream().toObservable(), (repo1, repo2) -> repo1)

subscribe

repo1.getStream()

Observable

repoObservable

Single

repoSingle

repoObservable

repoSingle

Observable

repoObservable.getStream().zipWith(repoSingle.get().toObservable(), (BiFunction<User, User, Object>)(user, user1) -> {
   updateUiWithUser(user); <-- should be from the 
repoObservable.getStream()
   showMessage();
}

repoObservable.observeOn(AndroidSheculers.mainThread())
.subscribe(...)
repoSingle.observeOn(AndroidSchedules.mainThread())
.subscribe(...)

Observable<Long> clock = Observable.interval(500, TimeUnit.MILLISECONDS);
Subject<Long, Long> subjectLong = PublishSubject.create();

clock.subscribe(subjectLong);

itemsObservable.delay { triggerObservable /* Observable<Unit> */ }

triggerObservable

AtomicBoolean

Completable

PublishSubject

init

fun <T> Observable<T>.foo(): ConnectableObservable<T> {
  return someOperators...
         .replay(1)
         .someMoreOperators...
}

Cache

Cache<K, Observable<T>>

replayingShare()

Single

val successObservable: Observable<SomeType>
val errorObservable: Observable<AnotherType> // AnotherType can be mapped into a Throwable

request.flatMap { 
    // First event of successObservable OR errorObservable. successObservable -> just(event), errorObservable -> error(error)
}

zip

take(1)

startsWith(none)

purchaseFlowStore
            .getTransaction(transactionId)
            .takeUntil { it.state == "done" || it.state == "error" || it.state == "authorized" }
            .doOnError {
                logger.debug("TransactionInfo error! $it")
            }
            .repeatWhen { Observable.timer(5, TimeUnit.SECONDS) }
            .viewModelSubscription({
                logger.debug("TransactionInfo is ${it.state}")

                when (it.state) {
                    "done", "authorized" -> {
                        getOrder()
                    }
                    "error" -> {
                        ShowRequestFailedDialog()
                    }
                }
            }, {
                logger.debug("TransactionInfo error! $it")
            })

Observable.interval(2000, TimeUnit.MILLISECONDS)
                .repeatUntil {
                    System.currentTimeMillis() - startTime > 10000
                    true
                }
                .subscribe {
                    purchaseFlowStore
                        .getTransaction(transactionId)
                        .viewModelSubscription({
                            when (it.state) {
                                "done", "authorized" -> {
                                    getOrder()
                                }
                                "error" -> {
                                    showErrorDialog()
                                }
                            }
                        }, {
                            showErrorDialog()
                        })
                }

doWork

class Foo {

	private val sem = Semaphore(permit = 1)

	fun doWork(): Completable = 
      Completable.fromCallable { sem.acquire() }
        .andThen { doWorkInternal() }
	    .doOnError { sem.release() }
        .doOnComplete { sem.release() }
}

Single.just