These are chat archives for ReactiveX/RxJava

Aug 2018
Aug 07 2018 14:00
Hello people, is there a way with RXJava2 to have a cancellationToken to cancel a single ? What I’m trying to achieve is a long series of actions, at some point the user want to cancel it, so I want to stop my series of actions, but to cancel my actions I need to excecute some asynchronous task and I don’t want my single to finish until the cancellation is done. Is such a thing possible and how ? I have no clue how to implement this :/
Ignacio Baca Moreno-Torres
Aug 07 2018 15:43
Un-subscribe the single when the cancelation is done?
David Stocking
Aug 07 2018 23:39

@jaumard Ok so first I don't know if this is a great idea. I am just going to assume you have your reasons why just unsubscribing isn't good enough. Now on to how you would do this. If you wanted a cancellation token like C#, you could do it fairly easily if you used Observable.create. C# cancellation tokens are glorified boolean holders anyway. You could also use Maybe.create, Flowable.create, etc. I don't think you could do this with Single unless you just say it was an error on cancel. What you would do is in your tight loop you just check your cancellation token. My example will be in Kotlin (sorry I am primarily a Android programmer)

Observable.create { emitter ->
    for (val action : actions) ->
        if (token.cancelled()) {

Depending on what you are trying to do this might work. If you have a chain of Maybes, you could also just decide to not return the next step

fun Maybe<T>.ifNotCancelled(defered: () -> Maybe<R>): Maybe<R> {
    return this.filter { !token.isCancelled() }

    .ifNotCancelled { step2() }
    .ifNotCancelled { step3() }
    .ifNotCancelled { step4() }

There are quite a few ways to do this and without more info I am just taking shots in the dark, but I hope that help gives you an idea of what you can do. The reason this is a little strange is probably because you don't want to model something as part of the stream. Usually it is just easier to keep all the information as part of the stream then you can do whatever you want via more RxJava operators like filter, flatMap, map etc. One of the ways we do this in my project is with switchMap. When you don't stay inside the stream, your kind of just using RxJava as a glorified Functional library / Thread replacement which isn't horrible but kind of leaves all these great ideas behind.