Operators Skip, SkipLast, Take with time #667
Conversation
|
RxJava-pull-requests #601 FAILURE |
| private static final class SourceObserver<T> implements Observer<T>, Action0 { | ||
| final Observer<? super T> observer; | ||
| final Subscription cancel; | ||
| final AtomicInteger state = new AtomicInteger(); |
There was a problem hiding this comment.
Why do you need an atomic state machine in here when there is not going to be any concurrency when the Observer is invoked?
The concurrency will happen in a very controlled place when the timer fires and you emit whatever is queued and onCompleted, but the on* events will not be invoked concurrently.
There was a problem hiding this comment.
Take doesn't queue anything. It relays events until the timer fires, which might run at the same time as a regular onNext event. The atomic is there to prevent this.
There was a problem hiding this comment.
That's a simple compareAndSet then on being finished is it not?
There was a problem hiding this comment.
If I use the classic get/compareAndSet, the following case might happen:
T1: if (state.get()) succeeds
T2: if (compareAndSet(true, false)) succeeds
T1: observer.onNext() executing
T2: observer.onCompleted executing
One would need to mutually exclude observer.onXXX calls, which could be done via synchronization and some overhead.
Manual Merge of Pull #667
Rebased version, without the drain scheduler variant.