diff --git a/rxjava-core/src/main/java/rx/Observable.java b/rxjava-core/src/main/java/rx/Observable.java index 50fdfdc338..4349750baa 100644 --- a/rxjava-core/src/main/java/rx/Observable.java +++ b/rxjava-core/src/main/java/rx/Observable.java @@ -6900,9 +6900,20 @@ public final Subscription subscribe(Subscriber observer) { if (isInternalImplementation(observer)) { onSubscribeFunction.call(observer); } else { - onSubscribeFunction.call(new SafeSubscriber(observer)); + // assign to `observer` so we return the protected version + observer = new SafeSubscriber(observer); + onSubscribeFunction.call(observer); } - return hook.onSubscribeReturn(this, observer); + final Subscription returnSubscription = hook.onSubscribeReturn(this, observer); + // we return it inside a Subscription so it can't be cast back to Subscriber + return Subscriptions.create(new Action0() { + + @Override + public void call() { + returnSubscription.unsubscribe(); + } + + }); } catch (OnErrorNotImplementedException e) { // special handling when onError is not implemented ... we just rethrow throw e;