Proposal: Lifted Observable

[trxcllnt]

Implement Observable operators in terms of lift.

What is lift?

lift is a function that takes a source Observable and an Observer factory function, and returns a new Observable:

function lift (Observable source, Function observerFactory) {
    return new Observable((destinationObserver) => {
        return source.subscribe(observerFactory(destinationObserver));
    });
}

Though this is the formal definition, we should make the following changes to our Observable's lift:

1. Assign lift on the Observable prototype.
2. Remove the source argument in favor of using this as the source.
3. Change the type of the observerFactory from function to the ObserverFactory interface. This interface requires the implementor to have a create function that accepts an Observer and returns an Observer.

lift has the following advantages:

1. A reduction in the Observable subscription call-stack.
2. Completely factors out transient closures from Observable creation and subscription.
3. Operations performed on Observable subclasses can return the sub type if the implementor overrides lift on their Observable subclass.

For example, here's how map is implemented in terms of lift:

class Observable {
  constructor(subscribe) {
    if(subscribe) {
      this.subscribe = subscribe;
    }
  }
  subscribe(observer) {
    return this.source.subscribe(this.observerFactory.create(observer));
  }
  lift(observerFactory) {
    const o = new Observable();
    o.source = this;
    o.observerFactory = observerFactory;
    return o;
  }
  map(selector) {
    return this.lift(new MapObserverFactory(selector));
  }
}

class MapObserverFactory {
  constructor(selector) {
    this.selector = selector;
  }
  create(destination) {
    return new MapObserver(destination, this.selector);
  }
}

class MapObserver {
  constructor(destination, selector) {
    this.selector = selector;
    this.destination = destination;
  }
  next(x) {
    return this.destination.next(this.selector(x));
  }
  throw(e) {
    return this.destination.throw(e);
  }
  return(e) {
    return this.destination.return(e);
  }
}

[xgrommx]

I think that operator looks like as flatMap

[benlesh]

@xgrommx really the one you're interested in here would be mergeAll. flatMap is just a specialization of map then mergeAll.

[benlesh]

Per a discussion yesterday with @jhusain, @trxcllnt, @benjchristensen and others:

Advantage: Custom Types That Compose

This lift method will allow users to create custom observable types that "compose through" by overriding lift:

class PaulsCustomObservable extends Observable {
  lift(observerFactory) {
    const o = new PaulsCustomObservable();
    o.source = this;
    o.observerFactory = observerFactory;
    return o;
  }
}

Advantage: Operator Observer Decoration

By overriding lift in a custom type, or monkey patching from a script (for debugging purposes), it becomes possible to inject behaviors into all observers in an operator chain... for example:

(rudimentary example)

const _lift = Observable.prototype.lift;
Observable.prototype.lift = function(observerFactory) {
  return _lift.call(this, new LogObserverFactory(observerFactory));
};

class LogObserverFactory {
  constructor(actualObserverFactory) {
    this.actualObserverFactory = actualObserverFactory;
  }
  create(destination) {
    return new LogObserver(this.actualObserverFactory.create(destination))
  }
}

class LogObserver {
  constructor(destination) {
    this.destination = destination;
  }
  next(x) {
   log('nexting ' , x);
    return this.destination.next(x);
  }
  throw(e) {
    log('throwing ', e);
    return this.destination.throw(e);
  }
  return(e) {
     log('returning', e);
    return this.destination.return(e);
  }
}

Clearly more can be done to identify the destination observer, and possibly even log a stack frame with an Error object or something, but this is to get a general idea. This will be useful in my effort at Netflix to create an OSS debugging tool for RxJS (and RxJava).

Advantage: Fewer Classes?

It seems, on the surface at least, that there will be fewer classes involve in operators. The current (what's in master) version of RxJS Next creates an Observable and an Observer pair for each operation. With the lift method there will only be one Observable type for the most part. That should optimize much better in V8. Only perf tests will prove that out, though.

[benlesh]

I'd like to put this on the Roadmap immediately, and implement it very soon since it's key to forward progress. I'll leave this issue open for community discussion though.

[calebboyd]

This method seems very sound. I think this would be great! And definitely an improvement on what has been suggested before.

[benlesh]

@trxcllnt FYI: I have this conversion nearly completed in another branch... PR eminent.

[trxcllnt]

@blesh beautiful. looking forward to reviewing it.

[headinthebox]

Like! as well.

[staltz]

Advantage: Operator Observer Decoration By overriding lift in a custom type, or monkey patching from a script (for debugging purposes), it becomes possible to inject behaviors into all observers in an operator chain...

This advantage made me see how lift is going to be a really good thing.

PS: nitpicking on the code you probably copy-pasted from the MapObserver:

  next(x) {
    log('nexting ' , x);
    return this.destination.next(this.selector(x)); // this.selector(x) should be just x
  }

---

@blesh

really the one you're interested in here would be mergeAll. flatMap is just a specialization of map then mergeAll.

I see map and mergeAll as a specialization of flatMap. ;) Anyway, flatMap doesn't have the debugging and developer experience benefits that lift does.

[benlesh]

PS: nitpicking on the code you probably copy-pasted from the MapObserver:

haha.. oops. Edited.

[staltz]

Quoting @headinthebox in a discussion I had with him:

If you look at the old implementation of “first class events” in F#, you will see that they implemented every operator with a subject. For example (in pseudo syntax, and pseudo/broken code): Filter(p,xs) = { val s = new Subject<T>(); xs.subscribe(x => if(p(x) s.onNext(x)); return s; }

Subjects themselves are not a good idea for the implementation of lift, but I'm puzzled why lift should be implemented like this:

  lift(observerFactory) {
    const o = new Observable();
    o.source = this;
    o.observerFactory = observerFactory;
    return o;
  }
  map(selector) {
    return this.lift(new MapObserverFactory(selector));
  }

when it could be implemented like this

lift(observerTransform: (outObserver: Observer) => (inObserver: Observer)) {
  const o = new Observable(outObserver => this.subscribe(observerTransform(outObserver)));
  o.source = this;
  return o;
}
map(selector) {
  return this.lift(outObserver => {
    const inObserver = Observer.create(
      function next(input) {
        outObserver.next(selector(input));
      }
    );
    return inObserver;
  });
}

The above doesn't depend on the next(value : any) : any proposed change on the Observer interface. It can remain a sink as next(value: any) : void.

[trxcllnt]

@staltz because closures are too expensive.

[staltz]

Alright, that explains a lot. And what about the return of next(value : any) : any? It doesn't seem necessary to return here:

class MapObserver {
  // ...
  next(x) {
    return this.destination.next(this.selector(x));
  }
  // ...
}

[benlesh]

@staltz in fact, when this repository first started, we experimented with two implementations of observable: 1. The Observer/Observable pairing, and 2. Lift (as you've proposed it above) ... 1 was orders of magnitude faster. Mostly because it avoided closure. We've arrived at where we are (in master) with perf tests. Once this lift branch is fully functional, we'll perf test, and perf wins. (I'm sure it will be the same or better than master, though)

[benlesh]

Done in master.

[EmmanuelOga]

I may be wrong but the lift method reminds me of Dart's transform method.

[trxcllnt]

@EmmanuelOga yep, pretty much identical

[lock[bot]]

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.