forEach should unsubscribe if callback function throws

[zenparsing]

Test case:

new Rx.Observable(sink => {
    sink.next(1); 
    sink.next(2); 
    sink.complete();
}).forEach(x => {
    console.log(x);
    throw new Error("boom");
}).catch(err => console.log(err));

Expected result: Only "1" is logged.
Actual result: "1" and "2" are logged.

I recently fixed this issue over on es-observable:

• Spec (3.2.1 step 3.d)
• Reference Implementation

[benlesh]

Ah... good catch, thank you @zenparsing!

[benlesh]

Actually, @zenparsing, looking at your reference implementation, there's a tricky bit your not covering... consider this:

const syncObservable = new Observable((observer) => {
   const expensiveThing = new ExpensiveThing();
   observer.next('I am a little teapot!');
   return () => expensiveThing.destroy();
});

syncObservable.forEach(x => {
  throw new Error('dammit, I am a sugar bowl');
});

In the case above, the error will be thrown synchronously before subscription can be returned. So your reference impl will throw on this line

I haven't tested this, but I think that's what will happen.

I'm working up a solution for RxJS 5 now.

[zenparsing]

Yeah, I was trying to simplify the algorithm by letting the error propagate in synchronous case, but clearly I messed it up : )

[trxcllnt]

@zenparsing we had to go to quite a bit of trouble to ensure that both a. the source unsubscription logic is run and b. the error is thrown to outside consumers when a next/error/complete handler throws on synchronous dispatch.

[lock]

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.