forEach does not unsubscribe appropriately when next handler throws on synchronous emissions

[benlesh]

The current implementation is not properly unsubscribing from the source when the next handler throws in this scenario:

of(1, 2, 3).pipe(
  tap(console.log),
)
.forEach(() => {
  throw new Error('test error');
})
.catch(err => console.log(err.message));

// Logs:
1
2
3
"test error"

// Would expect just:
"test error"

Stackblitz here

[benlesh]

This is so glaring to me, I'm shocked it's not been reported. I guess it's an edge case, really. How often do people throw errors in the next handler of a forEach on a synchronous observable? But still. It's CRAZY to me that this wasn't caught before. And even worse: I didn't catch it while updating the code, and it passed review! haha.

Well, at least it's caught now.