RFC: Building operators the way **we** build operators.

[benlesh]

(NOTE TO READERS: I have some spikes already in-flight on this, just calling that out so no one wastes time trying to implement it)

Summary Of Proposal

• Add a finalize handler to Observer<T> so that all consumer subscriptions can utilize it.
• Add a second options argument to Observable#subscribe with an owner property that accepts:
  • a Subscription that controls the resulting subscription (similar to a cancelation token)
  • OR a Subscriber that not only controls the subscription, but also acts as a "destination" for creating operators.

Problem space

1. Currently, the way we recommend operators are developed and the way we actually develop operators inside of the RxJS project differ substantially.
2. The way we recommend people develop operators have some corner case issues with synchronous inner observables.
3. Exposing how RxJS creates operators internally is ugly and hard to read.

Different approaches externally and internally

Externally we tell people do basically do this:

export const myOperator = <T>(...args) => (source: Observable<T>) => new Observable(subscriber => {
 // do stuff here
 return source.subscribe({
   next: () => { },
   error: (err) => subscriber.error(err),
   complete: () => subscriber.complete()
 })
});

Internally, we're doing something like this:

export const rxjsOperator = <T>(...args) => operate((source, subscriber) => {
  source.subscribe(
    // function names left out to illustrate how hard to read it is.
    createOperatorSubscriber(subscriber, () => {}, () => {}, undefined, () => {}); 
  )
})

We would never want to productize that for general consumption. It's nonsense for the most part. You can see we're using the argument order to "know" what is what. This is done for size optimization mostly. The arguments are in this order: parent subscriber (the observation and subscription we want to chain), a next handler override, a complete handler override, an error handler override, and a finalization function (behaves like the finalize operator, in a sense).

The reason we needed to do things differently

The problem comes from a corner case where an operator that is subscribing to "inner observables" happens to subscribe to an inner observable that is synchronous, and then have another operator down the chain that wants to terminate the resulting subscriptoin.

Consider this scenario:

source.pipe(
  concatMap(() => {
    return range(0, Infinity).pipe(
      tap(console.log)
    )
  }),
  take(10)
)
.subscribe()

If concatMap was implemented naively, as we currently tell external users of RxJS to implement custom operators, they'd lock up their thread.

This is because range(0, Infinity) is a "synchronous firehose", meaning it will synchronously emit a LOT of values (an infinite number, in fact). And it should be okay, if the concatMap implementation connected it's subscriptions in advance, because the take(3) would close the subscription and end the emission of values from that infinite range. However, if the subscriptions aren't chained in advance, it will terminate the final subscription, but the range(0, Infinity) will continue to loop and try to emit values.

This is because of an inherent disadvantage to subscribe returning Subscription: It has to wait for the subscribe call to finish its work in order to return. Which means you're not able to cancel that child subscription until after you have a handle to it... A handle you will never get.

RxJS gets around this problem by providing "destination" subscriber to a newly created operator subscriber. Subscribers are the marriage of a subscription and the observer. When you provide a destination that is a subscription, the subscriptions are automatically chained.

For a variety of reasons, this API has not really been exposed in a meaningful, documented way for public consumption. But we want to make sure people can develop their operators in a way that will always work properly. It's also a bonus if they can use OUR operators as a reference because they're implemented in the same way.

Proposal: Change RxJS to create operators the same way as everyone else

This work is already underway for version 8. Very soon, you'll start to see that reflected. Basically all operators will be made with something like:

export const myOperator = <T>(...args) => (source: Observable<T>) => new Observable(subscriber => {
 // do stuff here
 return source.subscribe({

 })
});

However, there are a few things we'd need to make sure we were able to do with the public API cleanly before this could work.

1. We need to add a finalize handler to our Observer<T> that Observable#subscribe() takes.

I think this will be a welcome addition. This is necessary to make it more ergonomic and terse to run clean up for operators. Looking through our code base will show how this is needed. Basically what this would do will be fire when the subscription is finalized.

Basically this:

source.subscribe({
  // fires after "complete" or "error" or "unsubscribe".
  finalize: () => console.log('finalized!')
})

2. Allow a second configuration argument to subscribe to pass a parent subscription

This will be great for us in a lot of ways. Since in version 8, we only have one argument to subscribe, which must be either a function (next handler), or an object (the observer), that frees us up to have a configuration/options object as the second argument.

We want to do a configuration/options object so we can have named arguments, and non-breaking flexible API to add/change features in that area in the future.

This parent subscription would effectively be a cancellation token (and it's controller) in a way, but also, it would provide for us an API to chain Subscriptions and Subscribers.

If a Subscription is passed as the owner

This means that Subscription is the "parent" and owns the created/returned subscription. When the owner is unsubscribed, the created subscription is also unsubscribed. In effect, this will work like a cancellation token, but will provide no other semantic difference to current subscription.

If a Subscriber is passed as the owner

This means that the subscriber is the "parent subscription" now (Subscriber is both an Observer and a Subscription).

It ALSO means that the observer passed to subscribe will act like an override or interceptor for passing values to that owner Subscriber. For an example of how this will work, have a look at OperatorSubscriber in the current code base.

Safety

For the time being, these things will be determined by instanceof checks, to ensure we're dealing with compatible instances of Subscription or Subscriber. If the Subscriber or Subscription passed to the owner property of the subscribe options is not an instanceof Subscriber or instanceof Subscription, a TypeError will be thrown immediately. The only reason this could happen is if a developer is mixing instances from multiple bundled copies of RxJS, or if they're trying to literally pass the wrong object.

Alternatively, we may look into using a Symbol to "get" a known Subscriber or Subscription from whatever is being passed as an owner. This warrants investigation.

Proposed API

interface Observer<T> {
  next(value: T): void;
  error(err: any): void;
  complete(): void;
  finalize(): void; // NEW!
}

interface SubscriptionOptions {
  owner: Subscription | Subscriber<any>
}

interface Observable<T> {
  subscribe(
    handler: Partial<Observer<T>> | ((value: T) => void),
    options?: SubscriptionOptions // NEW!
  ): Subscription
  
  // (omitting other methods, etc)
}

General use as a cancellation token

And usage in a general, public sense could be like so to get an cancellation token effect (similar to an AbortSignal):

const owner = new Subscription();

a$.subscribe(console.log, { owner })
b$.subscribe(console.log, { owner })

// later unsubscribe from both:
owner.unsubscribe();

Use for chaining subscribers (in operators)

This is clearly a contrived example, but here

export function myOperator<T>(...args: any[]) {
  return (source: Observable<T>) => new Observable(
    subscriber => {
      source.subscribe({
        next: (value) => {
           // override next here
           subscriber.next(value + 1)
        },
        complete: () => {
           // override complete here
           subscriber.next('this is about to complete');
           subscriber.complete();
        }
      }, {
        owner: subscriber
      })
    }
  );
}

Use Cases

Just as an example here are a before and after of a simple custom "map" operator:

RxJS 5.5 - 7.x:

/**
 * BEFORE
 */
export function customMap<T, R>(project: (value: T) => R) {
  return (source: Observable<T>) => new Observable<R>(subscriber => {
    return source.subscribe({
      next: (value) => {
        let result;
        try {
          result = project(value)
        } catch (err) {
          subscriber.error(err)
          return;
        }
        subscriber.next(result)
      },
      error: (err) => subscriber.error(err),
      complete: () => subscriber.complete()
    })
  })
}

RxJS 8.x (proposed):

/**
 * AFTER
 */
export function customMap<T, R>(project: (value: T) => R) {
  return (source: Observable<T>) => new Observable<R>(subscriber => {
    source.subscribe(
      (value) => subscriber.next(project(value)), 
      { owner: subscriber }
    )
  })
}

Alternatives Considered

Alternative 1: The start event with Subscription

The start or subscribe handler on Observer that supplies the Subscription as soon as the subscription starts. This one isn't too bad, because we already have some precedent with a subscribe handler in tap. There's also some preceding design discussion in the TC39 observable proposal where this was considered, and I remember some of those conversations.

It looks like this:

source.subscribe({
  start: subscription => {
   // do whatever you need to do with the subscription here.
  },
  next: console.log,
})

// Or to be more in-line with our `tap` API:

source.subscribe({
  subscribe: subscription => {
   // do whatever you need to do with the subscription here.
  },
  next: console.log,
})

This solves the problem of needing to get access to the subscription before anything is emitted from the source... However it's gross and/or doesn't solve for everything:

1. There would then be a pattern of needing to "leak" the subscription out of that handler, with the knowledge that the act of calling subscribe on the source is going to synchronously trigger that prior to anything.
2. It can't possibly "auto-wire" the passthroughs to the destination subscriber with error handling. That means that all try/catch and error: err => subscriber.error(err) type boilerplate would still need to be done.

[demensky]

Maybe it's time to think about integration with AbortSignal? I want to give examples that will show the possibilities that integration with AbortSignal will give. The main advantage, in my opinion, is more transparent integration with the native API.

---

Subscription will not happen:

const signal = AbortSignal.abort();

range(0, Infinity).subscribe({
  next: (value) => {
    console.log('next', value)
  },
  signal,
});

---

Limit infinity range:

const controller = new AbortController();

range(0, Infinity).subscribe({
  next: (value) => {
    if (value >= 5) {
      controller.abort();
    }

    console.log('next', value);
  },
  signal: controller.signal,
});

---

Custom operator:

export function myOperator<T>(...args: any[]) {
  return (source: Observable<T>) => new Observable(
    subscriber => {
      source.subscribe({
        next: (value) => {
          subscriber.next(value + 1)
        },
        complete: () => {
          subscriber.next('this is about to complete');
          subscriber.complete();
        },
        error: (error) => {
          subscriber.error(error);
        },
        signal: subscriber.signal
      })
    }
  );
}

---

Better integration with native APIs:

// Polling 10 seconds, last fetch will be cancelled
defer((signal) => fetch('/foo', {signal}).then((response) => response.json()))
  .pipe(repeat(), takeUntil(timer(10_000)))
  .subscribe((result) => {
    console.log(result);
  });

source$
  .pipe(
    switchMap((body, signal) =>
      // the previous fetch will be correctly canceled on each new call to the handler
      fetch('/foo', {body, signal}).then((response) => response.json()),
    ),
  )
  .subscribe((result) => {
    console.log(result);
  });

---

fromEvent without TeardownLogic and removeEventListener:

function fromEvent<T>(target: EventTarget, eventName: string): Observable<T> {
  return new Observable<T>((subscriber) => {
    target.addEventListener(
      eventName,
      (event) => {
        subscriber.next(event as T);
      },
      {signal: subscriber.signal},
    );
  });
}

---

Integration with Node.js:

import type {FileChangeInfo} from 'node:fs/promises';
import {watch as nodeWatch} from 'node:fs/promises';
import type {Observable} from 'rxjs';
import {defer, takeUntil, timer} from 'rxjs';

function watch(file: string): Observable<FileChangeInfo<Buffer>> {
  return defer((signal) => nodeWatch(file, {signal}));
}

watch('foo.txt')
  .pipe(takeUntil(timer(200)))
  .subscribe((info) => {
    console.log('info', info);
  });

[benlesh]

@alexcastillo I fully agree. There's also another problem which is that it's unergonomic and (currently) slow as hell when it comes to adding and removing listeners.

[benlesh]

Also: I don't want the configuration to be the same as the observer (as it is in the examples). It must be separate. There are two different concerns there, and it's not good to mix them. Aside from that, if it were implemented this way and you wanted to subscribe with a Subject, you'd have to ... mix the subject with another object with an owner on it? Or temporarily set an owner on the Subject? It just doesn't work.

[alexcastillo]

Agreed. My vote goes to keeping RxJS runtime-agnostic.

[voliva]

As a note, one can very easily write a user-land operator to work with signals which can even be composed from defer:

const signaled = getPromise => defer(() => {
  const controller = new AbortController();
  return from(getPromise(controller.signal)).pipe(
    finalize(() => controller.abort())
  )
});

return signaled((signal) => fetch('/foo', {signal}));

[benlesh]

Also, at the point that we do decide to support signals, we can leverage the same options argument:

source.subscribe(fn, { signal }); // non-breaking addition at some point.

[voliva]

I've been kinda naughty in some of my projects and I've been using new Subscriber() (marked as internal by RxJS) to circumvent the original problem. In most of the cases it's not really needed, so whenever possible I really try to avoid using it, but there are cases where I can't find any other solution that would work.

I'm wondering how will the proposed API help avoid using new Subscriber(). One of the places we use it is in a very custom variation of a shareReplay, which I can try and remove all extra complexity and reduce it to a possible implementation of a share, just for this discussion:

const someReallyCustomShareThatDoesMoreThingsThanThisImplementation =
  <T>() =>
  (source$: Observable<T>) => {
    let subject: Subject<T> | null = null;
    let sourceSubscriber: Subscriber<T> | null = null;
    let refCount = 0;

    return new Observable<T>(subscriber => {
      refCount++
      let innerSub: Subscription

      subscriber.add(() => {
        refCount--;
        innerSub.unsubscribe();
        if (refCount === 0) {
          sourceSubscriber?.unsubscribe();
          subject?.complete();
          subject = null;
          sourceSubscriber = null;
        }
      })

      if (!subject) {
        subject = new Subject<T>();
        innerSub = subject.subscribe(subscriber);
        sourceSubscriber = new Subscriber<T>({
          next: (value: T) => {
            subject.next(value)
          },
          error: (err: any) => {
            sourceSubscriber = null
            subject.error(err)
            subject = null
          },
          complete: () => {
            sourceSubscriber = null
            subject.complete()
            subject = null
          },
        })
        source$.subscribe(sourceSubscriber)
      } else {
        innerSub = subject.subscribe(subscriber);
      }
    })
  };

The test that breaks if we try to use user-land observer + returned value of source$.subscribe( is exactly a firehose test, where source$ is the firehose and after some multiple shared subscriptions and a few emissions the refCount drops to 0.

When using sourceSubscriber this is fine, because on refCount == 0 we call .unsubscribe() and the source$.subscribe(sourceSubscriber) picks that up and closes the subscription.

Without using new Subscriber( with the current API, I'm not really sure what's a possible fix for that test. It gets really tricky because the source$ subscription is being shared by multiple subscribers, and the unsubscription happens somewhere else (when refcount reaches 0)

Am I right thinking that this API will help solve this, by creating a Subscription and passing it as the owner of the inner source$.subscribe?

Essentially like this:

const someReallyCustomShareThatDoesMoreThingsThanThisImplementation =
  <T>() =>
  (source$: Observable<T>) => {
    let subject: Subject<T> | null = null;
    let ownerSubscription: Subscription | null = null;
    let refCount = 0;

    return new Observable<T>(subscriber => {
      refCount++
      let innerSub: Subscription

      subscriber.add(() => {
        refCount--;
        innerSub.unsubscribe();
        if (refCount === 0) {
          ownerSubscription?.unsubscribe();
          subject?.complete();
          subject = null;
          ownerSubscription = null;
        }
      })

      if (!subject) {
        subject = new Subject<T>();
        innerSub = subject.subscribe(subscriber);
        ownerSubscription = new Subscription();
        source$.subscribe({
          next: (value: T) => {
            subject.next(value)
          },
          error: (err: any) => {
            ownerSubscription = null
            subject.error(err)
            subject = null
          },
          complete: () => {
            ownerSubscription = null
            subject.complete()
            subject = null
          },
        }, {
          owner: ownerSubscription
        })
      } else {
        innerSub = subject.subscribe(subscriber);
      }
    })
  };

If so, I'm really happy of this change. It puts an end of having to use some RxJS internals to build some of these operators.

[josepot]

I really, really, really like the direction that this RFC is taking. However, I would like to go one step further and make owner a first-class citizen, therefore adding it into the Observer interface:

interface Observer<T> {
  next(value: T): void;
  error(err: any): void;
  complete(): void;
  finalize(): void; // NEW!
  owner: Subscription | Subscriber<any>; // NEW!
}

interface Observable<T> {
  subscribe(
    handler: Partial<Observer<T>>,
  ): Subscription
  subscribe(
    handler: (value: T) => void,
    owner?: Subscription | Subscriber<any>
  ): Subscription
  
  // (omitting other methods, etc)
}

We want to do a configuration/options object so we can have named arguments, and non-breaking flexible API to add/change features in that area in the future.

I think that this would be a mistake. Having a "hotchpotch" options object "just in case" can be very problematic IMO. The reason being that it can easily introduce breaking changes as backwards compatible changes. For instance, lets say that in the future a signal property gets added into the options Object (please, please, please don't, though 😅 🙏, but just for the sake of this argument, lets consider it). At a first glance, that looks like a non-breaking change. However, that is in fact a breaking change, because all those operators that have been written without taking the signal property into account won't work if they are mixed with operators that take the signal property into account. So, I would much prefer having a proper breaking change, so that all operators adjust to the new "signal" option, than delivering a breaking-change as a backwards compatible one.

[benlesh]

I don't think we can do that. Then if someone subscribes with a Subject for the purpose of multicasting, the subject would need to... temporarily patch an owner onto itself?

Or if the user wants to subscribe with a function, but also wants to provide a shared cancellation mechanism via the owner. Would they need to patch an owner property onto the function?

In the end, the observer and the configuration/options of the subscription (such as the owner) are two different things.

[voliva]

I think semantically it would make more sense to have the owner as a property of Observer, because the owner is there to notify when the source should stop streaming to the callbacks. It's something that feels like it should go hand-in-hand.

I would go one step ahead and just straight remove the second optional property.
The way I see it, this owner property is for building operators, so you'll probably be using the object signature anyways (.subscribe({ next, error, complete, owner })). If you're using the shorthand .subscribe(v => ...) you're probably not interested in owner, but for those cases you really want to pass it in, it's not that hard to change it to .subscribe({ next, owner }).

@benlesh's question about a Subject is interesting. I think the solution would be to have the Subject have the property owner that points to itself (not just temporarily), just to keep it consistent.
At the same time this raises a question: A Subject is both Observer + Observable, so with this proposal to include finalize() to Observer, does it mean that consumers can also call subject.finalize()? What would that mean?

And as a last comment, @josepot raised a very good point about object properties and backwards-compatiblity.
A very good example is the issue I raised a few months ago #7078 where tap(subject) stopped to work properly.
This happened because of a change that added the properties subscribe, unsubscribe, finalize to the tap observer, causing tap to call .unsubscribe() on the subject when there was an unsubscription, which kills the subject and prevents any resubscription from happening ever again (it throws an error that the subject was unsubscribed).
This seemed like a non breaking change, but turns out it wasn't, and it wasn't using external objects, but an object exported by RxJS (a Subject)

I'm not sure what's the best solution with the finalize() and how it plays with a Subject, but my vote so far would be to move owner to Observer, and not to add a config/options parameter at this moment.

[benlesh]

This just simply cannot happen. They're two separate concerns. Mixing them causes API issues.

// This doesn't make sense
const handler = () => {
  // do something
}
(handler as any)['owner'] = subscriber
source.subscribe(handler)

// This doesn't make sense either
const subject = new Subject()
subject.owner = subscription1 // ???
source1.subscribe(subject)
subject.owner = subscription2 // ???
source2.subscribe(subject)

Whereas this works fine:

// GOOD API: This works
source.subscribe(() => {
  // do something
}, { owner })

// GOOD API: This works
const subject = new Subject();
source1.subscribe(subject, { owner: subscription1 })
source2.subscribe(subject, { owner: subscription2 })

There's the observer which is a registration point for handlers.

Then there's subscription options, which at this point would just be an "owner".

[voliva]

Thanks for your reply - Sorry, maybe I'm not explaining myself correctly - Also, I don't want to add friction to this, because I really like this feature, but I think it's worth considering the option of moving the owner to the Observer.

From your examples:

const handler = () => { /* do something */ }
// If you want to pass an owner to the handler, you can just use the Partial<Observer> interface
source.subscribe({ next: handler, owner: subscriber });

// If all you care about is only the `next`, you still have the shorthand overload
source.subscribe(() => /* do something */);

// This is exactly the same behaviour as the 7.0 API. If you want to add an error case to the shorthand above,
// you wouldn't do `handler['error'] = error => /* do something */ ` either. You would just change it
// to `source.subscribe({ next, error })`.

const subject = new Subject();
// If you want to pass a different owner than the subject, you can still pass it as normal
source1.subscribe({
  next: v => subject.next(v),
  error: e => subject.error(e),
  complete: () => subject.complete(),
  owner: subscription1
})

It might be slightly inconvenient on these cases, but it's as descriptive as it gets. And I feel there are very little cases where you want to do source.subscribe(() => ...) where you also need to pass in an owner. Which again, if you need to, just change it to source.subscribe({ next: () => ..., owner: subscription })

From my point of view having the owner as part of the Observer would allow you to declare the lifetime of the observer at the same point you are declaring it, which I think it might make sense exploring. Whereas the original proposal an observer is just a passive object, and it only adds a synchronous cancelation mechanism to the .subscribe( function.

Edit - I just now realised.... Maybe this confusion comes from the term "Observer". I don't mean the observer parameter on .subscribe, but the Observer interface, which currently has the properties next, error complete. This suggestion is about adding owner (or another name for the same concept) to this interface.

[kolodny]

Overall I like it, I think we we may be able to make the DX slightly cleaner for custom operators by having some factory like so:

const operator = <
  TSubscriber = unknown,
  TSource = unknown,
  Args extends readonly unknown[] = unknown[]
>(
  callback: (
    subscriber: Subscriber<TSubscriber>,
    ...args: Args
  ) => Partial<Observer<TSource>>
): ((
  ...args: Args
) => (source: Observable<TSource>) => Observable<TSubscriber>) => {
  return (...args: Args) => {
    return (source: Observable<TSource>) => {
      return new Observable<TSubscriber>((subscriber) => {
        const observer = callback(subscriber, ...args);
        // Soon: source.subscribe(observer, { owner: subscriber });
        source.subscribe(observer);
      });
    };
  };
};

Usage on what you had before would look something like like this:

// Initial Proposed API
function myOperator<T>(...args: any[]) {
  return (source: Observable<T>) =>
    new Observable((subscriber) => {
      source.subscribe(
        {
          next: (value) => {
            // override next here
            subscriber.next((value as number) + 1);
          },
          complete: () => {
            // override complete here
            subscriber.next('this is about to complete');
            subscriber.complete();
          },
        },
        {
          owner: subscriber,
        }
      );
    });
}

// Using the operator function
const myOtherOperator = operator((subscriber, ...args: any[]) => ({
  next: (value) => {
    // override next here
    subscriber.next((value as number) + 1);
  },
  complete: () => {
    // override complete here
    subscriber.next('this is about to complete');
    subscriber.complete();
  },
}));

We can even get this typed fairly well:

const maxSizeValidateOperator = operator(
  (subscriber: Subscriber<boolean>, maxSize: number) => {
    return {
      next: (value: string) => {
        // override next here
        subscriber.next(value.length <= maxSize);
      },
      complete: () => {
        // override complete here
        subscriber.next(false);
        subscriber.complete();
      },
    };
  }
);

of('cat', 'person', 'dinosaur', 'goat')
  .pipe(maxSizeValidateOperator(5))
  .subscribe((value) => console.log({ value })); // value is boolean

of(1, 2, 3)
  .pipe(maxSizeValidateOperator(5)) // Error: 'number' is not assignable to 'string'
  .subscribe((value) => console.log({ value }));

of()
  .pipe(maxSizeValidateOperator('bad')) // Error: 'string' is not assignable to 'number'
  .subscribe((value) => console.log({ value }));

We can also have the factory be curried so you don't need to smoosh the subscriber with the args if that's something that we want to do. e.g.

const maxSizeValidateOperator = operator(
  (subscriber: Subscriber<boolean>) => (maxSize: number) => {
    return {
      next: (value: string) => {
        // override next here
        subscriber.next(value.length <= maxSize);
      },
      complete: () => {
        // override complete here
        subscriber.next(false);
        subscriber.complete();
      },
    };
  }
);

But I'm leaning towards the non curried version.

Link to stackblitz

[benlesh]

After some thought I'm pulling back from this proposal.

The primary reason is that it doesn't align with any other observable implementations available, and we just went through some efforts to try to make RxJS operators somewhat "observable agnostic". As in, if there's another simplified observable implementation or something lands in a platform, RxJS operators should "just work" with them.

[benlesh]

Closed for new proposal here: #7253