Improve Observer to remove Subscriber

[voliva]

This proposal is triggered from the discussion in the RFC #7243, which suggests adding owner: Subscription | Subscriber as an optional parameter to the Observable.subscribe method.

The main goal of this one is to solve the same problem as the original RFC, but at the same time standardizing how Observers expose their lifetime (Partial observers, Subscribers and Subjects).

Authors

This proposal is co-authored by @voliva and @josepot

Problem space

The current API of subscribe has two overloads: shorthand and partial observer:

// next handler
const subscription = source.subscribe((value) => {
  /* do something */
});

// Partial observer (each property is optional)
const subscription2 = source.subscribe({
  next: (value) => { /* do something */ },
  error: (error) => { /* do something */ },
  complete: () => { /* do something */ },
});

This has the problem that source.subscribe doesn't know the lifetime of the observer, in other words, the observer can't stop the subscription.

Once .subscribe returns the Subscription, then the consumer can call .unsubscribe( to stop it, but this might not be an option if source emits a large amount of values synchronously, which is what's called a "firehose" in the original RFC.

With the current API there's another place we deal with Observers: We get a Subscriber when creating a new observable:

const source = new Observable((subscriber) => {
  /* do something */
});

The Subscriber is both a Subscription and an Observer, in other words, it's an Observer that also has the information about its lifetime: It can notify the source when it should be disposed.

The problem is that Subscriber is an implementation detail, and new Observable leaks it. It's used internally to solve the firehose issue when creating operators, but it's something that consumers shouldn't really use.

On the other hand this also adds in confusion on what's the difference between an Observer and a Subscriber.

And lastly we also have another kind of Observer, which is a Subject.

A Subject is an Observable and an Observer, but also has a lifetime by its own. It's not fully a Subscription, but it does have an .unsubscribe() method that closes it.

In the past, a Subject had a property isStopped (or going further into the past, isUnsubscribed), but this was marked as internal and then removed for v8.0. This is just another observer lifecycle which has a different way of getting represented than the other cases.

The original RFC proposes to add an owner configuration option on the Observable.subscribe function that would allow the consumer to manage the lifetime of the subscription for synchronous values.

const owner = new Subscription();
const subscription = source.subscribe(
  (value) => {
    /* do something */
  },
  { owner }
);

const subscription2 = source.subscribe(
  {
    next: (value) => { /* do something */ },
    error: (error) => { /* do something */ },
    complete: () => { /* do something */ },
  },
  { owner }
);

And this helps consumers build operators that would work with a firehose source. However, this looks like a perfect opportunity to let Observers have a consistent way of exposing what is their lifetime, in a way that it solves the same problem that the original RFC fixes.

Proposal: add a property to Observers that exposes their lifetime

The Observer interface changes the interface with a new property with its lifetime:

interface Observer<T> {
  next: (value: T) => void;
  error: (error: unknown) => void;
  complete: () => void;
  owner: Subscription;
}

Observable.subscribe keeps the existing signature

interface Observable<T> {
  // Shorthand
  subscribe(
    handler: (value: T) => void
  ): Subscription;
  
  // Partial observer
  subscribe(
    observer: Partial<Observer<T>>
  ): Subscription;
}

// Shorthand, you can only pass in "next" function
observable.subscribe(value => /* ... */);

// Passing in a partial observer, you can specify an `owner` if needed.
observable.subscribe({
  next: value => /* ... */,
  owner: subscription
})

And all interfaces that extend Observer will also have the owner subscription - Subscribers and Subjects.

In fact, Subscribers can be removed, which has the advantage that the API won't have to leak an internal implementational detail.

new Observable(observer => {
  observer.next('a');
  observer.next('b');

  // subscriber.add(() => {}) becomes...
  observer.owner.add(() => {});

  // Subscribing to other observables will have everything wired up
  source.subscribe(observer); // <-- This would be a noop operator

  // Note how the owner is passed implicitly, which means that we don't
  // even need to return a teardown function. When the observable is finalized
  // `observer.owner` will be unsubscribed and the `.subscribe(` above will
  // stop the subscription.
});

// Or you can add any custom behaviour you want:
new Observable(observer => {
  source.subscribe({
    next: v => observer.next(v),
    error: v => {
      console.log('error detected oh no', v);
      observer.error(v)
    },
    complete: () => observer.complete(),
    owner: observer.owner
  });
});

// Or change the owner to have a different unsubscribe logic
// Implementing a take(5) here as an example
new Observable(observer => {
  let count = 0;
  source.subscribe({
    next: v => {
      observer.next(v)
      count++;
      if (count == 5) {
        observer.complete()
        // Another example of how this change would make subscription management
        // more composable and simple. This `observer.complete()` finalizes the
        // Observable, which causes `observer.owner` to become unsubscribed.
        // This will also stop the subscription on `source.subscribe` without
        // actually having to do anything else.
      }
    },
    error: v => observer.error(v),
    complete: () => observer.complete(),
    owner: observer.owner
  });
})

To ease migration, the original unsubscribe function could be added to the observer returned by the Observable constructor, which would just call this.owner.unsubscribe().

Lastly Subjects would also have their owner exposed, which is just a reference to their Subscription. This will make it consistent with the rest of the Observers, and it will let subjects with complex lifetimes to expose that in a consistent way.

The following example shows a share() implemented with this proposal, compared to the original RFC's API.

const userCreatedShare = () => source => {
  let subject = null;
  let refCount = 0;

  return new Observable(observer => {
    refCount++;

    observer.owner.add(() => {
      refCount--;
      if (refCount === 0) {
        // Because we have passed subject to `source.subscribe(subject)`,
        // this will unsubscribe from the source.
        subject.owner.unsusbscribe();
      }
    });

    if (!subject) {
      subject = new Subject();
      subject.owner.add(() => {
        // Put it back to null once it's disposed
        subject = null;
      });
      subject.subscribe(observer);
      source.subscribe(subject);
    } else {
      subject.subscribe(observer);
    }
  })
}

// Just as a comparison, this is what it would look like with the original RFC:
const userCreatedShare = () => source => {
  let subject = null;
  let ownerSubscription = null;
  let refCount = 0;

  return new Observable(subscriber => {
    refCount++;

    subscriber.add(() => {
      refCount--;
      if (refCount === 0) {
        subject.complete();
        // We can't use the subject.owner because it's not part of the Observer
        // so we need to handle it through an external subscription
        ownerSubscription.unsubscribe();
        ownerSubscription = null;
      }
    });

    if (!subject) {
      subject = new Subject();
      ownerSubscription = new Subscription();
      ownerSubscription.add(() => {
        subject = null;
      });

      // Need to manually pass in the owner
      subject.subscribe(subscriber, { owner: subscriber });
      source.subscribe(subject, { owner: ownerSubscription });
    } else {
      subject.subscribe(subscriber, { owner: subscriber });
    }
  })
}

Another advantage that moving lifetime management to owner on the subject, is that tap would work with a Subject again (it's currently broken because tap calls 'unsubscribe' #7078), because that action would now be moved to subject.owner.unsubscribe(). This is just another example of how having a consistent API often leads into good interop.

We are not completely familiar with the current internals of RxJS, but we really think this would be a very good addition, and would make the API consistent, simpler and less bug-prone to use.

We also agree this might be a big change, so it's possible that it might not be possible to implement it for the next major release (8.0), but we still think it could be used as part of a bigger roadmap across multiple releases.

cc @benlesh

[benlesh]

I'm sorry. This design is a hard "no" from me. For some of the same reasons I explained in the other thread here: #7243 (reply in thread)

I can see that you're trying to address those concerns, but I think the proposed API here is still clunky.

There's an additional problem with BOTH of our proposed solutions in that it means the RxJS operators are proprietary again, when I went to some lengths to try to make them observable agnostic in newer versions of 8.x.

I can tell that you're passionate about this solution, and you've put a lot of thought and time into it, and I'm so very sorry, but I'm actually going to close both this issue and the other for now. As I have a different proposal coming.

[benlesh]

New discussion around new propose is here: #7253

(I'm so sorry again, @voliva, and I really do appreciate all of the thought and effort you put in above)

[josepot]

Hi @benlesh ! Thanks for looking into this and for your reply. Looking forward to reading your new proposal. In the meanwhile, though, could you please elaborate on this:

There's an additional problem with BOTH of our proposed solutions in that it means the RxJS operators are proprietary again, when I went to some lengths to try to make them observable agnostic in newer versions of 8.x.

I really fail to understand how this proposal makes operators "proprietary again". Could you please explain what you mean by that? 🙏

[voliva]

Thank you for going through it and giving it a thought, but I still kindly disagree.

I don't fully understand why do you think the proposed API here is still clunky. Based on the reasons + examples from the original thread #7243, the only thing that this approach has vs the original RFC is that it's less convenient when you want to resubscribe as-is, but changing only a different owner. In that sense, both approaches are "equally clunky" if you have to do an operator that changes some of the next, error or complete callbacks.

I think I also understand the concern about "proprietary operators", in that those that are created using these APIs won't really work with other implementations of Observable. But the solution to this problem is really hard, because there's no standard on how the subscriptions/unsusbscriptions are managed on Observables.

Even one of your new proposals in #7253 would still be "proprietary operators", since subscriber.intercept( is a function that wouldn't be available in non-RxJS observables.

I just took a look now at the proposal of TC39 for Observables (still in stage 1, so it's subject to change), and the current proposal is something similar to what our proposal suggests, but I think more convoluted and less consumer-friendly: The interface Observer has a start(subscription: Subscription) function that will get called with the Subscription created when calling Observable.subscribe(observer). This should allow the observer to abort the subscription in the firehose case.

This is just to say that I'm not sure if the "proprietary operators" problem is even addressable. Every Observable implementation has a different way of managing their subscriptions, and having operators that work with all of them is going to be really challenging if there are no standards around that.

(Or maybe I missunderstood the proprietary terminology)

[josepot]

I think I also understand the concern about "proprietary operators", in that those that are created using these APIs won't really work with other implementations of Observable. But the solution to this problem is really hard, because there's no standard on how the subscriptions/unsusbscriptions are managed on Observables.

I don't think that's what @benlesh meant by "propietary operators". I really don't see how that would be the concern of the operators themselves... Specially taking into account the fact that moving forward Observables won't have pipe and there will be a new "pipe" function (rx) which takes an Observable(like) as a first argument, and a list of operators next. Therefore, that function could apply a from to its initial argument, so that the operators don't have to deal with those shenanigans themselves... I mean, unless you want for the operators to be 100% interoperable on their own with any other Observable(like) out there, which IMO would just be a pretty bad idea, then nope, I don't think that's what he meant.

I mean, but even in that was the case, then it would be "as simple" as converting the source Observable(like) into an Rxjs Observable. I don't see how that would have anything to do with the design of the Observable/Observer interfaces 🤷