Creating Observables in Scala

[samuelgruetter]

The constructors for Observables in Scala have multiple problems. I'd like to start a systematic discussion about this, hoping to find a good solution.

First, a list of use cases that we eventually want to cover:

• usecase01: "Observable.create, the mother of all factory methods": construct from an OnSubscribeFunc
• usecase02: an empty Observable
• usecase03: emit only 1 given element
• usecase04: emit a given (vararg)list of elements
• usecase05: emit all elements of an Iterable
• usecase06: emit all elements of a possibly infinite Iterable
• usecase07: emit the single item returned by a Future
• usecase08: wrap a Java Observable (Scala only)
• usecase09: emit only a given exception in onError
• usecase10: emit a range of integers
• usecase11: emit one asynchronously calculated element (a shorthand for constructing a Future and then applying usecase07

Note

• Depending on the approach, one constructor might cover several usecases, since I tried to split them up as much as possible.
• This list only contains "very important" constructors, which are eligible to be called Observable.apply. Others, which have names everyone agrees on, such as defer, never, interval, generate, etc, are not listed.

Now a first approach01 would be to use the same names as Java does:

/*usecase01:*/ public static <T> Observable<T> create(OnSubscribeFunc<T> func)
/*usecase02:*/ public static <T> Observable<T> empty()
/*usecase03:*/ public static <T> Observable<T> just(T value)
/*usecase04:*/ public static <T> Observable<T> from(T t1, ... T tN) /* 10 overloads */
/*usecase05:*/ public static <T> Observable<T> from(Iterable<? extends T> iterable)
/*usecase06:*/ /* not yet implemented */
/*usecase07:*/ public static <T> Observable<T> from(Future<? extends T> future)
/*usecase08:*/ /* NA */
/*usecase09:*/ public static <T> Observable<T> error(Throwable exception)
/*usecase10:*/ public static Observable<Integer> range(int start, int count)
/*usecase11:*/ /* not yet implemented */

We decided against this because this has problem01: Observable.apply is not used, so we don't exploit a nice feature of Scala.

That's why we implemented this approach02 (that's version 0.15.1):

/*usecase01:*/ def apply[T](func: Observer[T] => Subscription): Observable[T]
/*usecase02:*/ /* special case of usecase04 */
/*usecase03:*/ /* special case of usecase04 */
/*usecase04:*/ def apply[T](items: T*): Observable[T]
/*usecase05:*/ /* special case of usecase04, example: */ Observable(myList : _*)
/*usecase06:*/ /* not yet implemented, not even in Java */
/*usecase07:*/ /* not yet implemented */
/*usecase08:*/ def apply[T](observable: rx.Observable[_ <: T])
/*usecase09:*/ def apply[T](exception: Throwable): Observable[T]
/*usecase10:*/ def apply(range: Range): Observable[Int]
/*usecase11:*/ /* not yet implemented */

But this also turned out to have problems:

problem02: If I write this:

val o = Observable((observer: Observer[Int]) => { observer.onNext(42) /* no Subscription returned by mistake */ })

Then I don't get an error, but o is an Observable[Observer[Int] => Unit].

problem03: Observable(new Exception, new Exception) yields an Observable[Exception] with 2 elements and is not the same as Observable(new Exception) ++ Observable(new Exception), which yields an Observable[Nothing] with 0 elements, terminating with onError. Coursera students got confused about this.

problem04: The varargs apply and the OnSubscribeFunc apply clash in such a way that parameter type inference is lost:

val o1 = Observable(observer => { Subscription{} }) // Error: missing parameter type, should infer Observer[Nothing]
val o2 = Observable[Int](observer => { observer.onNext(1); Subscription{} }) // Error: missing parameter type, should infer Observer[Int]
val o3: Observable[Int] = Observable(observer => { observer.onNext(1); Subscription{} }) // works

problem05: Cannot easily construct an Observable emitting one Future, one Exception, or one Range.

problem06: It's possible to define both

def apply[T](items: T*): Observable[T] 
def apply[T](items: Iterable[T]): Observable[T]

but when I want to use it (eg Observable(List(1, 2, 3))), I get

ambiguous reference to overloaded definition, both method apply in object Observable of type [T](items: Iterable[T])rx.lang.scala.Observable[T] and method apply in object Observable of type [T](items: T*)rx.lang.scala.Observable[T] match argument types (List[Int])

We could also use implicit conversions, approach03:

• usecase01: "Observable.create": call it Observable.apply
• usecase02: List().toObservable
• usecase03: List(1).toObservable
• usecase04: List(1, 2, 3).toObservable
• usecase05: myIterable.toObservable
• usecase06: myIterable.toObservable
• usecase07: myFuture.toObservable
• usecase08: myJavaObservable.toObservable
• usecase09: Observable.error(new Exception)
• usecase10: (0 to 4).toObservable
• usecase11: Observable.async{ ... } or something else

Here, usecases 02, 03, 04, 05, 06, and 10 would all be covered by one single implicit conversion from Iterable[T] to Observable[T].
However, I'm not yet sure if this approach would lead to other problems.

I invite everyone to post new approaches, and to comment on existing ones. And please use increasing unique ids for usecase, approach and problem, to keep our discussion tidy ;-)

[samuelgruetter]

@benjchristensen @headinthebox @jmhofer @phaller @retronym @vjovanov @xeno-by and everyone else, your comments are welcome ;-)

[headinthebox]

Here is my 2 cents.

1. "Observable.create" (note the quotes people :-) is the mother of all constructors, so apply should be optimized for that.
2. Add as many other overloads as possible that do not interfere with 1)
3. Also have explicit names, even for create. Using extensive overloading on apply is IMHO not that great since you loose intent. For example say I want to do Observable.just(new Exception("As a regular value")) versus Observable.error[Integer](new Exception%28"Trigger onError"%29).

[samuelgruetter]

What about approach04:

/*usecase01:*/ Observable.create(observer => {...})
/*usecase02:*/ Observable()
/*usecase03:*/ Observable(oneValue)
/*usecase04:*/ Observable(1, 2, 3)
/*usecase05:*/ Observable.from(myIterable)
/*usecase06:*/ Observable.from(myInfiniteIterable)
/*usecase07:*/ Observable.from(myFuture)
/*usecase08:*/ /* internally: */ new Observable(javaObservable) /* from outside: */ toScalaObservable(javaObservable)
/*usecase09:*/ Observable.error(new Exception())
/*usecase10:*/ Observable.from(1 to 10) // covered by usecase05
/*usecase11:*/ Observable.from(Future{ ... }) // for the moment

Rationale:

• If everyone calls it Observable.create, why shouldn't we do so, too? And adding an Observable.apply which does the same is not a good idea: Two methods with different names which do the same, that's confusing people a lot.
• Reserving Observable.apply for varargs seems to be a waste at first sight, since it's not frequently used in "real world code". But it's very frequent in unit tests, and good code comes with many unit tests, so don't treat unit tests as second-class citizens. Additionally, writing documentation and reasoning about Observables becomes more lightweight: For instance, you can say "myFunc(Observable(1, 2, 3)) should return Observable(1, 4, 9)".
• Overloading from with Iterable and Future should not be a problem, since these types are disjoint.
• Observable.error: don't loose intent, as Erik says.

[retronym]

Overloading is less dangerous in Java because you have to pass type arguments explicitly.

Scala, on the other hand, will infer these.

IMO, this sort of API is broken:

def foo[A](as: A*)
def foo(a: X)

The user thinks they are calling the second method, gets something wrong so they types don't line up, and they are silently funnelled into the first. You are robbing the users of type safety.

[retronym]

BTW, I haven't been following this change closely enough to know if your proposal still falls afoul of my criteria, just take that as general advice...

[samuelgruetter]

thanks, approach01 and approach02 suffer from exactly this problem, but approach03 and approach04 should be fine

[samuelgruetter]

I think approach04 is the way to go, so I made a PR for it: #561

[samuelgruetter]

Quoting from an email by @headinthebox :

The main argument is for using Observable literals, as Samuel says

you can say "myFunc(Observable(1, 2, 3)) should returnObservable(1, 4, 9)",

However, in that case it is not much more verbose to say Observable(List(1,4,9)), especially since
in tests you would write assertEquals(List(1,4,9), xs.toBlockingObservable.toList) anyway.

Also, we can add extension methods such as List(1,2,3).toObservable().

Note that you should really not use Observable(List(...)) without a scheduler,
and the varargs overload does not allow for a scheduler, so it sets a bad example.
The easiest way to create an observable is also the "wrongest".

So, my proposal is to remove apply(args: T*) in favor of "create".

[...]

@Test def toObservable() {
  val xs = List(1,2,3).toObservable().toBlockingObservable.toList
  assertEquals(List(1,2,3), xs)

  val ys = Observable(List(1,2,3)).toBlockingObservable.toList
  assertEquals(List(1,2,3), xs)

  val zs = Observable(1,2,3).toBlockingObservable.toList
  assertEquals(List(1,2,3), xs)
}

[samuelgruetter]

I'm fine with replacing Observable(1, 2, 3) by Observable(List(1, 2, 3))+List(1, 2, 3).toObservable.

But I would not like to have two names for the same method: Observable.apply(Observer => Subscription) and Observable.create(Observer => Subscription), that's very confusing and bloats the (already large) API unnecessarily. We really should decide for either apply or create, but not both.

[samuelgruetter]

I'd prefer create ("don't loose intent"), but not having two methods which do the same is more important for me.

[headinthebox]

I don't mind having two at all. If we start with one we can always add apply(...) back, but if we put it in it is hard to remove.

[samuelgruetter]

Here's a trick by mentioned by @vjovanov which gives us more possibilities: Instead of having apply(T*), which makes apply unusable for any other overload, we could have just(T) and apply(T, T, T*).

[headinthebox]

I did seriously think about this option, it does effectively introduces two ways for the same thing, which is why I picked “items”.

On Dec 12, 2013, at 7:01 AM, samuelgruetter notifications@github.com wrote:

Here's a trick by mentioned by @vjovanov which gives us more possibilities: Instead of having apply(T_), which makes apply unusable for any other overload, we could have just(T) and apply(T, T, T_).

—
Reply to this email directly or view it on GitHub.

[samuelgruetter]

For the record, here's the what we currently have in master: approach05:

/*usecase01:*/ Observable.create(observer => {...})
/*usecase02:*/ Observable.empty
/*usecase03:*/ Observable.items(oneValue), List(oneValue).toObservable
/*usecase04:*/ Observable.items(1, 2, 3), List(1, 2, 3).toObservable
/*usecase05:*/ Observable.from(myIterable), myIterable.toObservable
/*usecase06:*/ // infinite Iterables not yet supported by RxJava
/*usecase07:*/ Observable.from(myFuture)
/*usecase08:*/ toScalaObservable(javaObservable)
/*usecase09:*/ Observable.error(new Exception())
/*usecase10:*/ Observable.from(1 to 10) // covered by usecase05
/*usecase11:*/ Observable.from(Future{ ... }) // for the moment

apply is unused here, but reserved for Observable.create. I think it's good to make one release without apply, so that everyone notices that there were changes in the constructors.