CODEWITHSUNDEEP
rx-java
Eugene Popovich
Eugene Popovich

Reputation: 3483

Rx Observable share() with skip() usage issue

I have a code, where the shared observable is used. One of the shared instances there uses extra skip and debounce operators. The problem is that sometimes the action for the instance where skip and debounce is used is not called

Observable<Integer> beginChanged = RxBindingUtils
        .valueChanged(begin)
        .doOnEach(value -> Timber.d("Begin value changed: " + value.getValue()))
        .share();

monitor(
        beginChanged
                .map(minutes -> minutesToTime(minutes))
                .subscribe(beginTime.asAction()));
monitor(
        beginChanged
                .map(minutes -> minutes / interval)
                .subscribe(rangeBegin.asAction()));
monitor(
        beginChanged 
                .skip(1)// skip initial value emitted automatically right after the
                        // subsription
                .debounce(500, TimeUnit.MILLISECONDS)// such as range bar may change the
                        // value very quickly use the
                        // debounce function for the timeout
                        // based processing
                .doOnEach(value -> Timber.d("Begin value changed 2: " + value.getValue()))
                .subscribe(mSchedulerRangeBegin.asAction()));

If i replace the last occurrence of the beginChanges with the new observable it works as expected

monitor(
        RxBindingUtils
                .valueChanged(begin) // can't use shared observable because of unexpected behaviour with skip call
                .skip(1)// skip initial value emitted automatically right after the
                        // subsription
                .debounce(500, TimeUnit.MILLISECONDS)// such as range bar may change the
                        // value very quickly use the
                        // debounce function for the timeout
                        // based processing
                .doOnEach(value -> Timber.d("Begin value changed 2: " + value.getValue()))
                .subscribe(mSchedulerRangeBegin.asAction()));

Are there any restrictions which don't allow to use skip and debounce on shared observables?

Upvotes: 1

Views: 276

Answers (1)

Eugene Popovich
Eugene Popovich

Reputation: 3483

Here is the answer from the akarnokd taken from github

Okay, I see what's wrong. The begin emits its current value on subscription which is triggered by the first Subscriber to share() and since all is synchronous, the subsequent subscribers don't get anything at that point because they are not yet subscribed. When the first event arrives, it gets skipped and only the second event is then processed. If you replace share with replay().refCount(), the test passes.

In my case the issue was solved by replacing share() with the publish() and after all the subscribers subscribed connect() call.

Upvotes: 1

Related Questions