Optimize Observable.concat and Observable.concatDelayError

[alexandru]

Currently we've got these functions on the companion object:

  def concat[A](sources: Observable[A]*): Observable[A] =
    Observable.fromIterable(sources).concatMap[A](identity)

  def concatDelayError[A](sources: Observable[A]*): Observable[A] =
    Observable.fromIterable(sources).concatMapDelayErrors[A](identity)

They are defined in terms of concatMap, however we can define them in terms of ++, the strict implementation (not the one with the by-name param, after #643 happens).

So the challenge, should you choose to accept it:

1. provide an implementation based on ConcatObservable instead of fromIterable and concatMap
2. prove that it is better by a benchmark, you can dump a quick benchmark in the benchmarks sub-project ... but don't commit it in the PR

[agajek]

I've tried to provide some implementation by stacking sources into ConcatObservable. It ended up with sth like that:

def concat[A](sources: Observable[A]*): Observable[A] =
    sources.toList.foldLeft(Observable.empty[A])((acc, a) => new ConcatObservable[A](acc, a))

In benchmark I've generated 3k Observable containing pure Int and then applied my implementation of concat to them. But the performance was terrible.
Current implementation is:

ObservableConcatBenchmark.concatObservable    3000  thrpt   20  139531619.493 ± 6637768.282  ops/s

and mine:

ObservableConcatBenchmark.concatObservable    3000  thrpt   20  12198.799 ± 1458.540  ops/s

Is the sth I need to put attention while implementing this?

[alexandru]

No longer relevant.