Single<T>.doOnAfterSuccess

Calls the action with the emitted value when the Single signals onSuccess. The action is called after the observer is called.

Source

subscribeOn

 freezes both its upstream source and downstream observer, all the Disposables (upstream’s and downstream’s) are frozen as well, all the values (including errors) are not frozen by the operator

Observable

upstream.map {  it.freeze() }

struct FooView: View {
  let presenter: FooPresenter

  var body: some View {
    Present(presenter) { model ->
      Text(model.name)
    }
  }
}

subscribeOn

observeOn

subscribeOn(ioScheduler)

private fun <T : Any> query(query: (TodoDatabaseQueries) -> Query<T>): Single<Query<T>> =
        queries
            .observeOn(ioScheduler)
            .map(query)

observeOn(mainScheduler)

subscribeOn(ioScheduler)

database
                .updates
                .observeOn(mainScheduler)
                .map(Result::ItemsLoaded)
                .subscribeScoped(onNext = ::dispatch)

//inside jsMain actual ChatClient
    private val observable : Observable<MessageModel> = 

    actual fun subscribeToNewMessages(observer or something to subscribe to the stream of messages){
        observable.subscribe With The Observer
    }
    actual fun somehow Clear the Subscription when Neccessary()

    socket.on("received_message") { mess ->
       observable.somehow Send to observers (mess)
    }

1.2.1

combineLatest

List

CoroutineExceptionHandler

coroutines-interop

Subjects

buffer

maybe

single

completable

fun listFilesFromFTP(ftpConnection : Connection) : Pair<Connection, List<Files>>

connectionObservable.flapMap(::listFilesFromFTP).flapMap(::filterLocalFiles)  ... and so on...

debounce

rxjava-interop

Combine

RxSwift

override fun observeAll(): Observable<List<RosterPresenceItem>> = queries.query(RosterDatabaseQueries::loadAll)
		.observe {
			it.executeAsList().map { e -> … }
		}

val updateSubject = PublishSubject<Unit>()

repeatWhen()

updateSubject.startWith(Unit).switchMap { database.fetch() }.sharedReplay(1)

coroutines-interop

@Composeable 
fun Foo() {
  val state by someReaktiveSingle.asRxJava2SingleSource().subscribeAsState()
  ...
}

commonMain

implementation("com.badoo.reaktive:rxjava3-interop:1.3.0")

compose-multiplatform-ios-android-template

2.2.0

2.0.0

toList