Is there a proper way to filter an Observable stream and log the ones that have been left out?

I see that ktor-client-ios has cancelation handler https://github.com/ktorio/ktor/blob/master/ktor-client/ktor-client-ios/darwin/src/io/ktor/client/engine/ios/IosClientEngine.kt Will it work with coroutine cancelation ? I mean if we wrap it by

singleFromCoroutine

and on dispose it will be canceled ?

Hello! Is there a native way to convert Reaktive classes into RxSwift?

Hi @Arkadii Ivanov, I noticed in v1.1.18 you made the wrappers types parameters non-null (SingleWrapper, ObservableWrapper, etc), this is now preventing me form doing

SingleWrapper<String?>

Is there any particular reason to this change? Or workaround to deal with Nullability/Absent of Values?

Hi @Arkadii Ivanov, with the latest changes in kotlin 1.4 with Ktor, I can see we are able to run it in background on native if we run from

MainScope

already tested that on my project, so would it make sense if this is provided in Reaktive through coroutine-interop?

Reaktive 1.1.19 is released! 📣 • Updated Kotlin to 

1.4.20

 and coroutines to 

1.4.2

 (#568) • Added test assertions chaining (#545) • Added 

invoke

 operator for 

ValueCallback

 and its 

Consumer

 typealias (#554) • Various performance improvements (#548, #547, #546) • Use 

abort()

 instead of 

exitProcess()

 in Kotlin/Native 

reaktiveUncaughtErrorHandler

 for better crash stack traces (#567) • Fixed a bug in 

Observable.buffer

 skip logic (#571) • Sources are now published for 

rxjava2-interop

 and 

rxjava3-interop

 modules (#560 by @CherryPerry) • Updated docs (#556)

Reaktive is now synchronized with JCenter! Now there should be no need for the additional Bintray repository definition

maven("<https://dl.bintray.com/badoo/maven>")

what would something like Traceur look like for reaktive?

Hi. Could you please elaborate what is getting frozen?

Hi @Arkadii Ivanov I see you merged kotlin 1.5.0 PR 4 days ago. Are there plans for releasing a new version soon?

Hi, is there a way to set an unhandled reaktive exception handler similar to Rx? I’m having trouble tracking down a native crash on iOS coming from CreateDefaultUncaughtErrorHandler and the stack trace doesn’t give much clues. abort() is called so it crashes the app. I’d like to set the default handler to log the error but not crash the app.

not sure if this is reaktive’s responsibility, but I just learned that this operator always calls

mapper

with the same list _instance_:

Iterable<Observable<T>>.combineLatest(mapper: (List<T>) -> R)

can it use a

CoroutineExceptionHandler

for launching coroutines?

Is it possible to recreate ReplayingShare with reaktive’s operators? https://github.com/JakeWharton/RxReplayingShare

Hi, I try ti use

Single<T>.doOnAfterSuccess

, as doc says

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

but my action is called first, before onSuccess in subscriber. what can be wrong?

Am I reading this correctly that reaktive does not freeze items emitted by a

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

I’m consuming an

Observable

from k/native, and I wanted to confirm that doing this is expected:

upstream.map {  it.freeze() }

I found a super simple way for using Reaktive streams with SwiftUI. Sharing it here in case it’s useful to others: https://gist.github.com/saket/62b1ccfc1a92c393be60a1cfd5357d98 Here’s what my usage code looks like:

struct FooView: View {
  let presenter: FooPresenter

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

Has anyone tried exposing a function to Swift for creating an observable without losing generics?

Hello, just started using Reaktive and have a general question of when to use

subscribeOn

and

observeOn

. When I was looking at the Todo sample I saw the repository had the following which if I understand correctly will make sure the values being observed are consumed on the background thread? Is there a reason to not also chain a

subscribeOn(ioScheduler)

to enforce all values computed are on a background thread? Or is that just SqlDelight only doing queries on a background thread that prevents the need for that?

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

I noticed that in one of the stores that

observeOn(mainScheduler)

is only called to make sure updates to UI are done on the UI, but there is no

subscribeOn(ioScheduler)

so was curious how the computations were being done on a background thread?

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

Thanks!

I have a ChatClient in my common code ( expect class currently implemented in JS) and am trying to have some kind of method to subscribe to new messages that come from the WebSocket. And to clear the subscription when the subscriber gets cleaned up. (Using stores from MVIKotlin). I am really lost at the moment, I cannot navigate the library to find what I need, this is what I am trying to achieve

//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)
    }

Hopefully this makes sense

Hi people, do you know any open source project with Ktor and Reaktive? It looks like Ktor is using MT coroutines now, I want to try the integration with Reaktive.

Reaktive

1.2.1

is released! •

combineLatest

 now emits 

List

 copies instead of same instances (#627) • Used 

CoroutineExceptionHandler

 in 

coroutines-interop

 module for better exception handling (#628) • Added KDocs for all operators and 

Subjects

 (#632, #633, #640, #642, #643) • Updated Kotlin to 1.5.30 (#637) • Added Apple silicon targets (#639) • Added various 

buffer

 operators (#645)

Hey everyone, I was wondering if there is a source of documentation other than the docs in the source code?

Hi! Do you have plans to update to kotlin 1.6.0?

hi, i am new to reaktive and Rx in general. i try to listen to server-sent events (with ktor cio) and i would like to know if any of the existing ways to create observables might fit my needs. i thought i could use the reaktive coroutine functions but since there was only

maybe

,

single

and

completable

, i ended up creating my own observable. it seems to work but i would like to know, if this is the way that is meant to be used. especially in combination with coroutines. (see thread for code)

Hello everyone! I don't have solid concepts on ReactiveX yet. What I'm trying to do is to download new files from an FTP server. The sequence that I'm currently thinking of is connect -> list files from FTP -> filter out files that are already in local path -> download new files. How to do this on a ReactiveX way, without having to pass the FTP connection as a parameter and returning it as an output for the next operator? I don't think having something like this is the correct way:

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

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

Hi everyone! I have subscription with

debounce

It works well on iOS, but on Android it doesn’t fire if app is in background near 15 min. There is some specific I should know ?

Hi everyone! Is there anyone working with Reaktive and JavaFX? I need a JavaFX Scheduler. Thanks!

Hi everyone! There is

rxjava-interop

What is about

Combine

or

RxSwift

? Did somebody try to implement it ?