1.x Take operator can emit more values than its limit

[mgp]

This can happen if a doOnNext action downstream can cause the observable upstream to emit another value. Here's a test that demonstrates this behavior:

@Test
public void testEmittingMoreValuesThanLimit() {
    final AtomicBoolean emittedFirstValue = new AtomicBoolean(false);
    final PublishSubject<Integer> subject = PublishSubject.create();
    final int firstValue = 1;
    final int secondValue = 2;

    // Record emitted values with this action.
    Action1<Integer> record = mock(Action1.class);
    InOrder inOrder = inOrder(record);

    subject
            .take(1)
            .doOnNext(record)
            .doOnNext(new Action1<Integer>() {
                @Override
                public void call(final Integer integer) {
                    // As the first value passes through, emit the second value.
                    if (!emittedFirstValue.getAndSet(true)) {
                        subject.onNext(secondValue);
                    }
                }
            })
            .subscribe();

    inOrder.verifyNoMoreInteractions();

    // Will record both values even though take(1) is upstream.
    subject.onNext(firstValue);
    subject.onCompleted();
    inOrder.verify(record).call(firstValue);
    inOrder.verify(record).call(secondValue);
    inOrder.verifyNoMoreInteractions();
}

I first noticed this in a project that uses RxJava 1.1.10. The onNext implementation of OperatorTake has changed a bit in master. Before I go ahead and submit a PR that fixes this behavior, I just wanted to ensure that it wasn't deliberate? Thanks!

[akarnokd]

Hi and thanks for discovering this. The behavior is not intentional. We are generally not prepared for synchronous reentry such as this. Solving it is a bit tricky because a delicate behavior needs to be ensured right at the limit. Here is how I'd fix this:

• I'd introduce a boolean stop variable on the parent Subscriber
• I'd change the onNext to have the reentrancy protection:

@Override
public void onNext(T i) {
    if (!isUnsubscribed()) {
        boolean stop = ++count >= limit;
        if (stop) {
            if (this.stop) {
                return;
            }
            this.stop = true;
        }
        child.onNext(i);
        if (stop && !completed) {
            completed = true;
            try {
                child.onCompleted();
            } finally {
                unsubscribe();
            }
        }
    }
}

This will prevent a recursive onNext call to be delivered after the limit.

[mgp]

Thanks for the response! I think alternatively you could do:

if (!isUnsubscribed() && (count++ < limit)) {
    boolean stop = count >= limit;
    child.onNext(i);
    if (stop && !completed) {
        completed = true;
        try {
            child.onCompleted();
        } finally {
            unsubscribe();
        }
    }
}

Which only changes the outer conditional and the computation of stop. LMK if that sounds good and I'll prepare a PR?

[akarnokd]

Looks even better, go ahead with the PR.

[akarnokd]

@mgp Are you going to post a PR against 1.x?

[akarnokd]

Closing via #3384.