I think this can end up being called concurrently which breaks the Rx contract.

I think `observer` needs to be wrapped by `SynchronizedObserver`.