How to get a one-time object for subscribers in RxJava2

Question

This may be a dumb question, but how do I get Disposable when subscribing to a Subject to Observable in RxJava 2.0?

For instance:

 observable.subscribeWith( behaviorSubject) 

doesn't return Disposable ? How to cancel such a subscription?

Or another example with CompositeDisposable :

 compositeDisposable.add( observable.subscribeWith( behaviorSubject) ) ) 

This does not compile because subscribeWith( behaviorSubject ) does not return one-time.

How to unsubscribe / delete / cancel correctly using Subjects?

Answer 1

You are not the first to stumble over this. For example, issue # 4438 . Just wrap your object with DisposableObserver . In another issue on Flowable someone suggested using one of the take*() methods to complete a subscription.