title: Combine multiple items from observable to new objects using RxJava
body: <p>I'm trying to implement some kind of streaming parser. Let's say that I have stream of integers, and I combine them to create new object, which aggregates part of stream.</p>
  
  <p>For example, object is "done" when integer is negative. To keep it simple, produced items will be strings of numbers.</p>
  
  <p>Here is simple example:</p>
  
  <pre><code>Source: 1, -2, 3, 4, -5, 6, -7, 8, 9, 10, -11
  Output: "1-2", "343-5", "6-7", "8910-11"
  </code></pre>
  
  <p>I can't find proper combination of existing operators. My solution was to provide custom operator, which aggregates data till "chunk" is valid and emits new object:</p>
  
  <pre><code>rx.Observable&lt;Integer&gt; src = rx.Observable
      .from(Arrays.asList(1, -2, 3, 4, -5, 6, -7, 8, 9, 10, -11));
  
  rx.Observable&lt;String&gt; res = src
      .lift(new rx.Observable.Operator&lt;String, Integer&gt; () {
  
          @Override
          public rx.Subscriber&lt;? super Integer&gt; call(
              final rx.Subscriber&lt;? super String&gt; subscriber) {
  
              return new rx.Subscriber&lt;Integer&gt;(subscriber) {
  
                  private final StringBuilder cur = new StringBuilder();
  
                  @Override
                  public void onCompleted() {
                      if (!subscriber.isUnsubscribed()) {
                          subscriber.onCompleted();
                      }
                  }
  
                  @Override
                  public void onError(Throwable e) {
                      if (!subscriber.isUnsubscribed()) {
                          subscriber.onError(e);
                      }
                  }
  
                  @Override
                  public void onNext(Integer i) {
                      if (subscriber.isUnsubscribed())
                          return;
  
                      cur.append(i).append('');
                      if (i &lt; 0) {
                          subscriber.onNext(cur.toString());
                          cur.delete(0, cur.length());
                      }
                  }
              };
          }
      });
  
  rx.observers.TestSubscriber&lt;String&gt; sub =
      new rx.observers.TestSubscriber&lt;&gt;();
  res.subscribe(sub);
  List&lt;String&gt; l1 = sub.getOnNextEvents();
  </code></pre>
  
  <p>It works well till I use some positional operators on them, like <em>take(N)</em> or <em>combineLatest</em>.</p>
  
  <pre><code>rx.Observable&lt;String&gt; res2 = res.take(2);
  </code></pre>
  
  <p>In this situation I don't get <em>onCompleted</em> or <em>onError</em> and smaller amount of items are emitted.</p>
  
  <p>It looks like each input <em>onNext</em> should have output <em>onNext</em> called.</p>
  
  <p>Any ideas?</p>