Observable.create may call source callback with already disposed emitter

[feerbach]

I’m observing behavior that seems strange to me when I use an Observable constructed in the following way.

I’m using Observable.using to create an observable based on some resource. In Observable.using, I pass a function as sourceSupplier, which returns an observable created using Observable.create.

Then I apply the takeUntil operator to the observable created by Observable.using to combine it with another observable — stopSubject, which in this example is a BehaviorSubject.

        class Resource {
            boolean isClosed = false;
        }

        BehaviorSubject<Object> stopSubject = BehaviorSubject.create();

        stopSubject.onNext(1);

        Observable
                .using(
                        Resource::new,
                        (resource) -> Observable.create(emitter -> {
                            ...
                        }),
                        (resource) -> {
                            resource.isClosed = true;
                        }
                )
                .takeUntil(stopSubject)

I expect that when the callback passed to Observable.create is invoked, the resource created by Observable.using has not yet been disposed, and the emitter passed to the callback is not yet disposed either. I verify this in the tests of the attached TakeUntilTest class.

TakeUntilTest.java.zip

These expectations hold true if I emit a value into stopSubject after subscribing to the observable under test (see the test otherObservableEmitsValueAfterSubscription).

However, if I emit a value into stopSubject before subscribing to the observable under test, the following happens:
The callback passed to Observable.create is invoked, but by that time the resource has already been disposed, and the emitter passed to the callback is already in the disposed state. This is verified in the test otherObservableHasValueInAdvance.

I looked into the RxJava implementation, and I believe the reason for this behavior is as follows:

ObservableTakeUntil class in its subscribeActual method first subscribes to the other observable and only afterwards subscribes to the source observable:

    @Override
    public void subscribeActual(Observer<? super T> child) {
        TakeUntilMainObserver<T, U> parent = new TakeUntilMainObserver<>(child);
        child.onSubscribe(parent);

        other.subscribe(parent.otherObserver);
        source.subscribe(parent);
    }

In my test, stopSubject acts as the other observable, and since it already has a value in advance, subscribing to it immediately triggers the TakeUntilMainObserver.otherComplete method, which writes a special DISPOSED value into the TakeUntilMainObserver.upstream field (using the helper method DisposableHelper.dispose):

        void otherComplete() {
            DisposableHelper.dispose(upstream);
            HalfSerializer.onComplete(downstream, this, error);
        }

The subsequent subscription to the source observable in ObservableTakeUntil.subscribeActual calls subscribeActual on ObservableCreate, which first calls onSubscribe on the passed observer, and only then invokes the source callback:

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        CreateEmitter<T> parent = new CreateEmitter<>(observer);
        observer.onSubscribe(parent);

        try {
            source.subscribe(parent);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            parent.onError(ex);
        }
    }

The call to observer.onSubscribe inside ObservableCreate ends up calling TakeUntilMainObserver. onSubscribe, which tries to set the passed disposable into the upstream field using the helper method DisposableHelper.setOnce:

        @Override
        public void onSubscribe(Disposable d) {
            DisposableHelper.setOnce(upstream, d);
        }

But since the upstream field already contains the special DISPOSED value, the assignment doesn’t happen, and instead DisposableHelper.setOnce calls dispose on the passed disposable. In this case, the disposable is an instance of ObservableUsing$UsingObserver, which proceeds to dispose the resource:

        @Override
        public void dispose() {
            if (eager) {
                disposeResource();
                upstream.dispose();
                upstream = DisposableHelper.DISPOSED;
            } else {
                upstream.dispose();
                upstream = DisposableHelper.DISPOSED;
                disposeResource();
            }
        }

Therefore, when control returns back to ObservableCreate.subscribeActual after onSubscribe is called, and the source callback is finally invoked, the resource has already been disposed, and the emitter passed to the callback (CreateEmitter) is already in disposed state.

I believe this current behavior is incorrect, and in the test otherObservableHasValueInAdvance, one of the following should happen:
• Either the callback passed to Observable.create should not be invoked at all,
• Or the callback should be invoked before the resource is disposed and the emitter is transitioned to the disposed state.

[akarnokd]

The current implementation works as intended.

Either the callback passed to Observable.create should not be invoked at all

Some callback may want to still run when the resource is disposed. You may need to check the disposed state within the callback since in the general case, dispose can happen asynchronously to the execution of the body.

Or the callback should be invoked before the resource is disposed and the emitter is transitioned to the disposed state.

Not possible, because the callback may be something blocking thus the control may never return to the subscribeActual. Simple example: a callback which sleep-waits in a loop to detect disposing the emitter, which would deadlock.

[feerbach]

Not possible, because the callback may be something blocking thus the control may never return to the subscribeActual. Simple example: a callback which sleep-waits in a loop to detect disposing the emitter, which would deadlock.

I didn’t know that Observable.create supports blocking callbacks. The documentation for this method states:

Provides an API (via a cold Observable) that bridges the reactive world with the callback-style world.

The documentation also provides the following example:

Observable.<Event>create(emitter -> {
    Callback listener = new Callback() {
        @Override
        public void onEvent(Event e) {
            emitter.onNext(e);
            if (e.isLast()) {
                emitter.onComplete();
            }
        }

        @Override
        public void onFailure(Exception e) {
            emitter.onError(e);
        }
    };

    AutoCloseable c = api.someMethod(listener);

    emitter.setCancellable(c::close);

});

If it is allowed for the ObservableOnSubscribe callback to be invoked with an already disposed emitter, then it suggests adding a check at the start of the callback:

Observable.<Event>create(emitter -> {
    if (emitter.isDisposed()) {
        return;
    }
    ...
});

You may need to check the disposed state within the callback since in the general case, dispose can happen asynchronously to the execution of the body.

However, if in the documentation example the api is created using Observable.using, then checking only emitter.isDisposed() is not enough:

        Observable.using(
                Api::new,
                (api) -> Observable.<Event>create(emitter -> {
                    Callback listener = new Callback() {
                        // ...
                    };

                    AutoCloseable c = api.someMethod(listener);

                    emitter.setCancellable(c::close);
                }),
                (api) -> {
                    api.close();
                }
        );

Since the api may be closed during the execution of the ObservableOnSubscribe callback, access to the api needs to be synchronized:
before calling api.someMethod(listener), the callback must ensure that the api hasn’t been closed yet.

As a result, the code becomes more complex. And all this complexity arises because Observable.create tries to support blocking callbacks.

In this situation, I feel it might be easier to write my own replacement for ObservableCreate, which would call onSubscribe on the provided observer after invoking the ObservableOnSubscribe callback. This way, all these explicit checks inside the callback could be avoided.

Do you think this is the right solution?

[akarnokd]

Yes.