RxJS stop foreach when observable meets condition

Question

I have an array of objects. I have to check each object trough an API call to see if the object is eligible for a certain promo. What is the best way to keep calling the API with objects until the last object is called in which the observable should return false, or one of the objects get a true state from the API?

At the moment I have this as code but it feels like there should be a better way with RxJS operators.

checkProductForPromo(items: []) {
   const promoChecker$ = new EventEmitter<boolean>();

   items.forEach((item, key, arr) => {
        // This does a HTTP callback to the API
        this.ApiService.getProductInformation(item).subscribe(
            (product) => {
                // Based on some properties this will return true or false of the product is eligible for the promo.
                if (product.isEligibleForPromo()) {
                    promoChecker$.emit(true);
                } else if (Object.is(arr.length - 1, key)) {
                    promoChecker$.emit(false);
                }
            }
        );
    });

    return promoChecker$;
}

Answer 1

You can create an Observable with the from operator:

checkItemsForPromo(items: any[]) {
  return from(items).pipe(
    concatMap(item => this.ApiService.getProductInformation(item)),
    map(product => !!product.isEligibleForPromo()),
    takeWhile(isEligible => !isEligible, true)    
  );
}

This Observable calls the API for each item in order. It waits for the previous request to complete before sending the next one. The map operator sets the emission to true or false as required.

The takeWhile means the Observable will continue to emit false until a true value comes along, at which point it will complete. The true argument passed to takeWhile is in fact the inclusive flag, meaning that the Observable will emit the single true value before completing.

If you only want it to emit false once on completion as opposed to for each non-eligible product, try adding a distinctUntilChanged() to the pipe.

Don't forget to subscribe to the Observable returned by this method, and unsubscribe if and when necessary.