I'm trying to reuse an Observer for a Single and Observable stream by creating only a single instance of DisposableSingleObserver/DisposableObserver and passing them through it through the subscribeWith() method on the stream like below:
public class SomeClass {
private DisposableSingleObserver<Object> observer;
public SomeClass() {
observer = new DisposableSingleObserver<Object>() {
@Override
public void onSuccess(Object object) {
...
}
@Override
public void onError(Throwable throwable) {
...
}
};
}
public void doSomeStuff() {
singleStream.subscribeOn(...)
.observeOn(...)
.subscribeWith(observer);
}
}
The above code resulted in ProtocolViolationException when I tried to subscribe with the single observer instance multiple times with message:
io.reactivex.exceptions.ProtocolViolationException: It is not allowed to subscribe with a(n) com.package.name.SomeClass$1 multiple times. Please create a fresh instance of com.package.name.SomeClass$1 and subscribe that to the target source instead.
So I modified the code a bit like below:
public class SomeClass {
...
public void doSomeStuff() {
if (observer != null) {
observer.dispose();
}
singleStream.subscribeOn(...)
.observerOn(...)
.subscribeWith(observer);
}
}
When I executed the above code, the ProtocolViolationException was no longer thrown and I was able to get event from the single stream successfully. But when I tried to call doSomeStuff() method couple of time, it completed successfully on first call but did not emit any event on second call. I was able to confirm that the subscription happened through doOnSubscribe() both times but the code emitting event for single stream was never executed on second call. So I have three questions:
- Why was the
ProtocolViolationExceptionthrown in the first place? - Why was I able to get the event on first subscription but not on the second subscription even though we subscribed to the source in both cases?
- How can I reuse observer by using only a single instance?