Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Proposal: Lifted Observable #60

Closed
trxcllnt opened this issue Jul 10, 2015 · 18 comments
Closed

Proposal: Lifted Observable #60

trxcllnt opened this issue Jul 10, 2015 · 18 comments
Assignees

Comments

@trxcllnt
Copy link
Member

@trxcllnt trxcllnt commented Jul 10, 2015

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

This comment has been minimized.

Copy link

@xgrommx xgrommx commented Jul 10, 2015

I think that operator looks like as flatMap

@benlesh

This comment has been minimized.

Copy link
Member

@benlesh benlesh commented Jul 10, 2015

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

@benlesh

This comment has been minimized.

Copy link
Member

@benlesh benlesh commented Jul 10, 2015

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

This comment has been minimized.

Copy link
Member

@benlesh benlesh commented Jul 10, 2015

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 comment has been minimized.

Copy link

@calebboyd calebboyd commented Jul 11, 2015

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

@benlesh benlesh self-assigned this Jul 12, 2015
@benlesh

This comment has been minimized.

Copy link
Member

@benlesh benlesh commented Jul 12, 2015

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

@benlesh benlesh mentioned this issue Jul 12, 2015
1 of 1 task complete
@trxcllnt

This comment has been minimized.

Copy link
Member Author

@trxcllnt trxcllnt commented Jul 12, 2015

@Blesh beautiful. looking forward to reviewing it.

@headinthebox

This comment has been minimized.

Copy link

@headinthebox headinthebox commented Jul 12, 2015

Like! as well.

@staltz

This comment has been minimized.

Copy link
Member

@staltz staltz commented Jul 13, 2015

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

This comment has been minimized.

Copy link
Member

@benlesh benlesh commented Jul 13, 2015

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

haha.. oops. Edited.

@staltz

This comment has been minimized.

Copy link
Member

@staltz staltz commented Jul 16, 2015

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

This comment has been minimized.

Copy link
Member Author

@trxcllnt trxcllnt commented Jul 16, 2015

@staltz because closures are too expensive.

@staltz

This comment has been minimized.

Copy link
Member

@staltz staltz commented Jul 16, 2015

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

This comment has been minimized.

Copy link
Member

@benlesh benlesh commented Jul 16, 2015

@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 ordered 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

This comment has been minimized.

Copy link
Member

@benlesh benlesh commented Jul 17, 2015

Done in master.

@benlesh benlesh closed this Jul 17, 2015
staltz added a commit to staltz/RxJSNext that referenced this issue Jan 4, 2016
Add tests for Observable to verify that the most important codebase-wide features brought by the
lift architecture are satisfied.

Refers to discussion in ReactiveX#60.
staltz added a commit to staltz/RxJSNext that referenced this issue Jan 13, 2016
Add tests for Observable to verify that the most important codebase-wide features brought by the
lift architecture are satisfied.

Refers to discussion in ReactiveX#60.
staltz added a commit to staltz/RxJSNext that referenced this issue Jan 18, 2016
Add tests for Observable to verify that the most important codebase-wide features brought by the
lift architecture are satisfied.

Refers to discussion in ReactiveX#60.
staltz added a commit to staltz/RxJSNext that referenced this issue Jan 26, 2016
Add tests for Observable to verify that the most important codebase-wide features brought by the
lift architecture are satisfied.

Refers to discussion in ReactiveX#60.
benlesh added a commit that referenced this issue Jan 27, 2016
Add tests for Observable to verify that the most important codebase-wide features brought by the
lift architecture are satisfied.

Refers to discussion in #60.
@EmmanuelOga

This comment has been minimized.

Copy link

@EmmanuelOga EmmanuelOga commented Mar 3, 2016

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

@trxcllnt

This comment has been minimized.

Copy link
Member Author

@trxcllnt trxcllnt commented Mar 3, 2016

@EmmanuelOga yep, pretty much identical

@lock

This comment has been minimized.

Copy link

@lock lock bot commented Jun 7, 2018

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.

@lock lock bot locked as resolved and limited conversation to collaborators Jun 7, 2018
Sign up for free to subscribe to this conversation on GitHub. Already have an account? Sign in.
Projects
None yet
Linked pull requests

Successfully merging a pull request may close this issue.

None yet
7 participants
You can’t perform that action at this time.