title: Exception in observable pipe is suppressed
body: <p>Suppose I want observable to emit value periodically until another observable emits. So I can use <code>timer</code> and <code>takeUntil</code> to achive that.
  But then I want to process every emitted value and stop (error) emitting when some condition becomes true. So I write next piece of code:</p>
  
  <pre class="lang-js prettyprint-override"><code>const { timer, Subject } = 'rxjs';
  const { takeUntil, map } = 'rxjs/operators';
  
  const s = new Subject();
  let count = 0;
  function processItem(item) {
    count++;
    return count &lt; 3;
  }
  
  const value = "MyValue";
  
  timer(0, 1000).pipe(
        takeUntil(s),
        map(() =&gt; {
          if (processItem(value)) {
            console.log(`Processing done`);
            return value;
          } else {
            s.next(true);
            s.complete();
            console.error(`Processing error`);
            throw new Error(`Stop pipe`);
          }
        }),
      )
  </code></pre>
  
  <p><a href="https://rxviz.com/v/7JXZKygo" rel="nofollow noreferrer">Playground</a></p>
  
  <p>But instead of getting error I have my Observable completed. 
  Only if I comment out <code>takeUntil(s)</code> operator, I get error.
  Looks like when pipe operator completes, it's value is not emitted immediately, but remembered and emitted at the end of next "iteration" of the pipe, then replaced by new result and so on. And in my situation next iteration, when error should be emitted, is prevented by <code>takeUntil</code>. Question is am I right with that assumption, and if so, why rxjs is designed in that way?</p>