operator request: groupByOrdered

[borissmidt]

The operator def groupByOrdered[K](key: T => K): Observable[(K, Observable[T])] would be handy to check a stream until it changes.
You could see it as a repeated takeWhile while equivalence of K is used to check if the condition changed.

The same behavior can be made with a scan, but i quite often ends up as complex code because you have to filter out the intermediate values and have to wrap everything in an option to know the end of the stream.

The groupByOrdered would also make it easier to reduce by key more efficiently if you know your data comes in a known order.
For example when you query a database.

def reduceByKeyOrdered[T, K](key : T => K)(op: (T, T)) ={
    o.groupByOrdered(key).flatMap { case (_, group) =>
      group.reduce(select)
    }
}

Of course a more generic implementation could accept (previous: T, current: T) => Boolean to know when to break the current group.

I wouldn't mind to implement this operator if you think it is a good idea.