title: rxjava consumes next element even after exception
body: <p>Could someone explain why rxjava consumes the next element from an observable sequence after exception in "onNext" happen and even after "onError" is called?</p>
  
  <p>Here is my simulation:</p>
  
  <pre><code>import java.util.concurrent.{TimeUnit, CountDownLatch}
  import rx.lang.scala._
  
  object Tests {
    val counter = new CountDownLatch(1)
  
    def buildStream(num: Int) = {
      Stream.range(1, num)
        .map {s =&gt;
        println(s"[${Thread.currentThread().getId}] Taken: $s");
        s}
    }
    val stream = buildStream(10).toSeq
    stream.toObservable
      .subscribe(
        onNext = x =&gt; consume(x),
        onError = e =&gt; println(s"ERROR! $e"),
        onCompleted = () =&gt; {println("completed"); counter.countDown()}
      )
  
   def consume(x: Int) = {
      println(s"[${Thread.currentThread().getId}] consuming: $x")
      Thread.sleep(100)
      if (x == 5) {
        throw new Exception(s"[${Thread.currentThread().getId}] consume?! ha!")
      }
    }
    counter.await(10, TimeUnit.SECONDS)
  }
  </code></pre>
  
  <p>and its result is the following, and you can see that item <strong>#6</strong> is taken from sequence after the exception is raised:</p>
  
  <pre><code>[39] consuming: 1
  [39] Taken: 2
  [39] consuming: 2
  [39] Taken: 3
  [39] consuming: 3
  [39] Taken: 4
  [39] consuming: 4
  [39] Taken: 5
  [39] consuming: 5
  ERROR! java.lang.Exception: [39] consume?! ha!
  [39] Taken: 6
  res0: rx.lang.scala.Subscription
  </code></pre>
  
  <p>As far as I can understand my observable collection created from a regular sequence is "cold" and the next item should only be taken from the source sequence after the current one is successfully handled by the observer.
  I could suspect some threading issue, but it would mean that "cold" is not that "cold" which cannot be true, also I can clearly see that the same thread ID is used all the time.</p>
  
  <p>So why item <strong>6</strong> is taken from the sequence?!</p>