every operator sending multiple values when re-entrant

[anthonyrota]

Describe the bug

The every operator can send multiple false's when, in the Subscriber's next, the Observable it is operating on sends a new value into the operator, because the middle subscriber's next handler will be recursively called before the complete is sent.

Related: #1967 #2100

  next: (value) => {
    if (!predicate.call(thisArg, value, index++, source)) {
      destination.next(false); # If inside this function call, the source sends a new value that the predicate rejects, two or more false's can be sent.
      destination.complete();
    }
  },

Also, it can send next(false) then next(true) then complete if re-entrant complete within the next commented.

Expected behavior

The circumstances leading to this shouldn't really happen but if it does, maybe consider guarding against it so only a single false is pushed?

Reproduction code

import { Subject } from 'rxjs';
import { every } from 'rxjs/operators';
{
  console.log('scen 1')
  const subj = new Subject();
  let i = 0;
  subj.pipe(every(() => false)).subscribe({
    next: (value) => {
      console.log('next', value, i);
      if (++i < 5) {
        subj.next(undefined);
      }
    },
    complete: () => {
      console.log('end');
    },
  });
  subj.next(undefined)
}
{
  console.log('scen 2')
  const subj = new Subject();
  subj.pipe(every(() => false)).subscribe({
    next: (value) => {
      console.log('next', value);
      subj.complete();
    },
    complete: () => {
      console.log('end');
    },
  });
  subj.next(undefined)
}

Reproduction URL

https://stackblitz.com/edit/rxjs-22nsn5?file=index.ts

Version

8.0.0-alpha.12

Environment

No response

Additional context

No response

[josepot]

This is clearly a bug, the observer should synchronously unsubscribed from the source as soon as it observes a value that doesn't meet the criteria.

If you were using v7, then you could use the following custom implementation which doesn't suffer from this bug:

const customEvery =
  <T>(cb: (value: T, idx: number, source: Observable<T>) => boolean) =>
  (source: Observable<T>): Observable<boolean> =>
    new Observable((observer) => {
      let index = 0;
      const subscriber = new Subscriber<T>({
        next(value: T) {
          if (!cb(value, index++, source)) {
            subscriber.unsubscribe();
            observer.next(false);
            observer.complete();
          }
        },
        error(e: unknown) {
          observer.error(e);
        },
        complete() {
          observer.next(true);
          observer.complete();
        },
      });
      source.subscribe(subscriber);

      return subscriber;
    });

I can't help to wonder what the official way of building a custom operator like that is for v8 🤔

EDIT:

I think that I found it. The new way of doing the equivalent in RxJS 8 would be:

const every =
  <T>(cb: (value: T, idx: number, source: Observable<T>) => boolean) =>
  (source: Observable<T>): Observable<boolean> =>
    new Observable((observer) => {
      let index = 0;
      const subscriber = operate<T, boolean>({
        destination: observer,
        next(value: T) {
          if (!cb(value, index++, source)) {
            subscriber.unsubscribe();
            observer.next(false);
            observer.complete();
          }
        },
        error(e: unknown) {
          observer.error(e);
        },
        complete() {
          observer.next(true);
          observer.complete();
        },
      });
      source.subscribe(subscriber);

      return subscriber;
    });

Example

props to @voliva for helping me out 😉