Ignore furthur messages after entering terminate state

[zsxwing]

As per #1633

[girishkolantra]

I have a question about implementing custom Subscribers. Should we always check for isUnsubscribed on the subscriber before doing further processing inside onNext onError and onCompleted? What's the best practice when implementing a custom Subscriber?

[zsxwing]

I have a question about implementing custom Subscribers. Should we always check for isUnsubscribed on the subscriber before doing further processing inside onNext onError and onCompleted? What's the best practice when implementing a custom Subscriber?

You don't need to check it. If you use subscribe(...), rxjava will wrap your subscriber with a SafeSubscriber which will check isUnsubscribed for you.

Usually, subscribe is enough. Some RxJava operators will use unsafeUnsubscribe(...) to avoid too many SafeSubscriber wrappers and improve the performance. But it's used only for implementing an Operator that requires nested subscriptions.

If you are not implementing an Operator, you should use subscribe.

[benjchristensen]

SafeSubscriber actually doesn't check isUnsubscribed. It filters out onNext after a terminal event (onError or onCompleted).

There are very few places where unsubscribe is triggered, such as take and that operator filters out anything received after unsubscribe because that is the job of that operator. If a Subscriber is invoking unsubscribe and is sensitive to race conditions it can check for isUnsubscribed but rarely is a user provided Subscriber doing this (generally something like take is doing it).

@girishkolantra What do you mean by "custom Subscriber"?

[girishkolantra]

I meant any subscriber that we implement. I have seen some sample code with checks isUnsubscribed and others that dont. I just wanted to understand what the differences are and when to use what. Thanks for taking the time to answer my questions.

[cloudbees-pull-request-builder]

ReactiveX » RxJava » RxJava-pull-requests #14 ABORTED

[zsxwing]

@girishkolantra we can continue the discussion here: #1633

[zsxwing]

@benjchristensen do I need to send another PR for 1.x?

[benjchristensen]

@zsxwing I think what we're really missing is unsubscribe like onErrorResumeNext has.

That said, it won't help to unsubscribe when inside a parallel since it doesn't really make sense to unsubscribe on an inner parallel Observable.

We may also want to add these same terminal state checks to onErrorResumeNext since a source that doesn't obey unsubscribe could still emit.

[mttkay]

@benjchristensen Why not add the check for isUnsubscribed in SafeSubscriber? I believe that might also fix this issue: #1590