Is it possible to implement an operator like delay but that also delays errors?

Question

I'm trying for some time now to implement an extension function (just becuse it's easier to me) that is capable of delaying both normal item emissions and errors. The existing delay operators only delays normal item emissions, errors are delivered ASAP.

For context, I'm trying to immitate an Android LiveData's behavior (kinda). LiveDatas are a observable pattern implementation that is lifecycle aware. Their observers are only notified if they are in a state where they can process that emission. If they are not ready, the emission is cached in the livedata and delivered as soon as they become ready.

I created a BehaviourSubject that emits the state of my Activities and Fragments when it changes. With that I created a delay operator like this:

fun <T> Flowable<T>.delayUntilActive(): Flowable<T> = delay { lifecycleSubject.toFlowable(BackpressureStrategy.LATEST).filter { it.isActive } }

and then use it like this

myUseCase.getFlowable(Unit)
                .map { it.map { it.toDisplayModel() } }
                .delayUntilActive()
                .subscribe({
                    view.displaySomethings(
                }, { }).addTo(disposables)

So even if myUseCase emits when the view is not ready to display somethings, the emission won't reach onNext() until the view does become ready. The problem is that I also want the view to displayError() when onError is triggered, but that too is lifecycle sensitive. If the view isn't ready, the app will crash.

So I'm looking for a way to delay both emissions and errors (onComplete would be good too). Is this possible? I tried some things with zip, onErrorReturn, delay inside delay, but nothing seemed right. I'd be equally unimpressed if this had a really easy solution I'm overlooking, or is impossible. Any ideas are welcome.

Bonus: any better way to do that for Single and Completable too? currently I'm just converting them to flowable.

Thanks in advance!

akarnokd akarnokd · Accepted Answer · 2019-08-13T15:10:05

You can handle the error via onErrorResumeNext, then taking the same error and delaying it via delaySubscription until your desired signal to emit said error happens:

source
.onErrorResumeNext({ error -> 
     Observable.error(error)
     .delaySubscription(lifecycleSubject.filter { it.Active } ) 
})

Is it possible to implement an operator like delay but that also delays errors?

1 Answers