What is the best way to interrupt async operation?

[NobodyLikesZergs]

Is it possible to interrupt operations started by Actor? In my case single Wish generate a chain of requests. How can i prevent execution of remaining requests when new Wish arrives?

[ShikaSD]

@NobodyLikesZergs
The simplest thing is to use takeUntil on your observable inside the actor.

class Actor(...) {
    private val interruptSignal = PublishSubject.create<Unit>

    override fun invoke(state: State, wish: Wish) = when (wish) {
        is Request -> {
            interruptSignal.onNext(Unit)
            dataSource.request(wish.data)
                .takeUntil(interruptSignal)
        }
    }
}

[NobodyLikesZergs]

@ShikaSD, Thanks a lot!

[zsoltk]

Probably we should add this to FAQ / best practices. Thanks for raising.

[qwert2603]

@ShikaSD @zsoltk
Could it be cleaner to create extra Actor class with signature (Observable<Pair<State, Action>>) -> Observable<Effect> ?

That will allow to cancel async operations like:

interface ActorObservable<State, Action, Effect> {
    fun invoke(input: Observable<Pair<State, Action>>): Observable<Effect>
}

sealed class Action {
    object Action1 : Action()
    object Action2 : Action()
}

data class Effect(val i: Int)

class ActorImpl<State> : ActorObservable<State, Action, Effect> {
    override fun invoke(input: Observable<Pair<State, Action>>): Observable<Effect> {
        return input
                .switchMap { (state, action) ->
                    when (action) {
                        Action.Action1 -> Observable
                                .interval(1, TimeUnit.SECONDS)
                                .map { Effect(1) }
                        Action.Action2 -> Observable
                                .interval(2, TimeUnit.SECONDS)
                                .map { Effect(2) }
                    }
                }
    }
}