Proposal: RxJava 2 - style Observer interface

[benlesh]

As discussed in the ES7 Observable Spec here, the idea looks like this:

interface Observer {
  subscribed((subscription: Subscription) => void): void
  next((value: any) => void): void
  error((err: any) => void): void
  complete(() => void): void
}

Advantages:

1. Actually knowing when a subscription starts (for things like subscribeOn)
2. Sets us up for backpressure handling later on.
3. Becomes trivial for end-user developer to add this.sub = subscription in the subscribed handler, giving a reference to subscription in all other handlers without closure.

[benlesh]

/cc @trxcllnt @benjchristensen

[trxcllnt]

@Blesh heh, this is identical to the Observer interface from my experiments this weekend. I approve!

[benlesh]

👍

[trxcllnt]

@Blesh one slight difference -- instead of a setter for subscription, my interface defined a public subscription property, which allows us to read the subscription from the "destination" observer and share it with the "source" observer. Check it:

interface IDisposable {
    isDisposed: boolean;
    dispose(): void;
}

interface ICompositeDisposable extends IDisposable {
    length: number;
    add(disposable: IDisposable | (() => void) | void): ICompositeDisposable;
    remove(disposable: IDisposable): ICompositeDisposable;
}

interface IObserver<T> extends IDisposable {
    disposable: ICompositeDisposable;
    next<T>(value: T): void;
    throw<E>(error: E): void;
    return(): void;
}

[benlesh]

@trxcllnt to further flush this out, the there is a Subscriber type that Rx Java uses that we're currently calling Observer or "safe" Observer:

class Subscriber implements Observer {
  subscribed((subscription: Subscription) => void): void
  next((value: any) => void): void
  error((err: any) => void): void
  complete(() => void): void

  // adds these
  add(subscription: Subscription): void
  unsubscribe(): void
  isUnsubscribed: boolean;
}

Then technically Observable ctor would get a Subscriber:

class Observable {
  constructor((subscriber: Subscriber) => Subscription|Function|void ): Subscription

  /** other stuff here **/
}

[headinthebox]

By far the easiest thing is to merge observer and subscriber (i.e. collapse both into observer). You never need to distinguish between them.

[benjchristensen]

This approach is where RxJava is heading with v2 as discussed here: ReactiveX/RxJava#2450

In particular, it will adopt the Reactive Streams API which are the result of 1+ year of collaboration with various companies. We explored many different improvements to the core Observable/Observer interface while adding backpressure, and the resulting onSubscribe(Subscription s) works elegantly, while also being efficient, which many alternatives (such as next returning a Future) were not. We have considered a Javascript definition: https://github.com/reactive-streams/reactive-streams-js.

The backpressure piece is not required for this to be a good design for RxJS, though I suggest it also does future-proof RxJS to allow that being added to support Node.js apps, which will end up needing it or all the same issues as we had with RxJava will result.

@trxcllnt the property model is great for a concrete implementation of IObserver. I suggest however that the interface be defined with onSubscribe(IDisposable) so that concrete implementations can behave however they wish.

// use term Subscription instead of Disposable so it represents the subscription lifecycle, and can be used for `request` semantics if that is added to RxJS
interface Subscription {
    isUnsubscribed: boolean;
    cancel(): void; // cancel/unsubscribe/whatever
    request(n: number); // optional support for micro-leasing to allow async flow-control, backpressure
}

interface Observer<T> {
    subscribed(subscription: Subscription); // always called when Observable is subscribed to, unless onError is called instead
    next<T>(value: T): void;
    throw<E>(error: E): void;
    return(): void;
}

[benlesh]

By far the easiest thing is to merge observer and subscriber (i.e. collapse both into observer). You never need to distinguish between them.

@headinthebox I agree. I guess what I'm trying to do is linguistically distinguish between "Observer The Interface" (as handed to subscribe) and "Observer: the safe thing we give to people building Observables" (as given in the Observable constructor) ?

[headinthebox]

There should only be one (and only one type of subscription)

trait Subscription {
  def unsubscribe(): Unit
  def isUnsubscribed: Boolean
  def +=(child: Subscription): Unit
  def -=(child: Subscription): Unit
}

trait Observer[-T] extends Subscription {
  def onNext(value: T): Unit 
  def onError(error: Throwable): Unit 
  def onCompleted(): Unit 
 }

[headinthebox]

@benjchristensen Why have two separate interfaces?

[benlesh]

@headinthebox I think we're all saying the same thing. The Subscriber implements both interfaces.

[headinthebox]

What I am saying is that we do not need a separate type for observer. Less abstractions == better abstractions.

[benjchristensen]

Why have two separate interfaces?

@headinthebox What are the separate interfaces you're referring to?

[benlesh]

What I am saying is that we do not need a separate type for observer. Less abstractions == better abstractions.

Yes... 100% agreed. I think that's where we're all heading.

[headinthebox]

@benjchristensen Observer and Subscription. Why doesn't Observer extend Subscription.