Incorrect observeOn main scheduler behaviour during unsubscribing

[benhongh]

observeOn operator seems to prematurely disconnect the observation pipeline during unsubscribing. To illustrate, let's start with a simple example:

let sequence: Observable<String> = create { obs in 
    return AnonymousDisposable { obs.on(.Completed) }
}

let subscription = sequence 
    >- subscribeCompleted { println("COMPLETED") }

subscription.dispose()

The output of the above shows "COMPLETED", which is the expected behaviour. However this is not the case if we observe the same sequence on the main scheduler:

let sequence: Observable<String> = create { obs in 
    return AnonymousDisposable { obs.on(.Completed) }
}

let subscription = sequence 
    >- observeOn(MainScheduler.shareInstance)
    >- subscribeCompleted { println("COMPLETED") } // does not get invoked

subscription.dispose()

It seems that after subscription.dispose() is called but before the AnonymousDisposable handler is called, the subscription chain is severed, so that the completed event never reaches the observer. I suspect the relevant line is in Producer.swift (line 36) where a CompositeDisposable is created and returned:

let d = CompositeDisposable(sink, subscription)

The order of the disposables passed into the CompositeDisposable may need to be reversed to ensure the custom disposable gets invoked first, but this is just my guess.

Ben

[kzaher]

Hi benhongh,

this has certainly started as an interesting morning :)

I could be wrong about this, so let me try to explain :)

The output of the above shows "COMPLETED", which is the expected behavior. However this is not the case if we observe the same sequence on the main scheduler:

Actually, I think what you are trying to do is undefined behavior :)

It is my understanding that sequence computation can complete in one of the following ways:

• it completes <- observer will receive .Completed
• it fails with error <- observer will receive .Error
• computation engine is disposed <- observer shouldn't receive ideally nothing

The third case it little complex because in various setup scenarios in multithreaded environment observer can receive additional events after dispose call has completed.

The reason why that is the case is because you could have a race between dispose process and produce process.

But on the other hand, just to make this clear, if you call dispose on one thread (like main), you won't observe any elements on that same thread. That is a guarantee.

The same is valid for serial dispatch queues.

Rx does have internal mechanisms that protect the system from rogue observables (the one that don't respect sequence grammar), but this is the first case I've seen of rogue disposable :)

The reason why I've labeled this disposable as rogue is because disposables shouldn't generate any elements as far as I know. Their responsibility is only resource disposal.

To illustrate this better, that would be like trying to unplug some electrical device, and in the process of unplugging it would always try to do something useful without power :)

I don't think that even Rx team considered that scenario :) But I could be wrong about this.

I could be maybe misunderstanding the problem, so could you please explain why do you feel that it should work that way?

Kruno

[kzaher]

Hi @benhongh ,

do you feel that your question has been answered and can I close this issue?

[kzaher]

Hi @benhongh ,

I didn't get any further feedback on this. I think it was a misunderstanding how Rx should work so I'm going to close this for now.