RFC: Developing operators the way *we* develop operators (part deux)

[benlesh]

Goals

• Make sure we're developing operators inside of RxJS in the same manner we'd tell others to develop operators.
• Make sure our operators remain somewhat "observable agnostic".
• Help address the "synchronous firehose" problem. (See explanation here)

Proposal

Productize our createOperatorSubscriber function such that it is easier to read and understand. This work would include:

• A better name (intercept?, chain?, hook? override? operate?).
• Making arguments named parameters.
• Possibly? Adding it as a method to Subscriber.

Something like this:

1. Basically the same as createOperatorSubscriber, only with a different name and named params:

export function myMap<In, Out>(project: (value: In) => Out): OperatorFunction<In, Out> {
  return (source) => new Observable(subscriber => {
    source.subscribe(intercept(subscriber, {
      next: value => subscriber.next(project(value))
    }));
  })
}

or...

2. Add a method to Subscriber that does this work:

export function myMap<In, Out>(project: (value: In) => Out): OperatorFunction<In, Out> {
  return (source) => new Observable(subscriber => {
    source.subscribe(subscriber.intercept({
      next: value => subscriber.next(project(value))
    }))
  })
}

---

EDIT: After some experimentation and some thought, I'm now leaning towards option 1 above. The reasoning:

• Option 1 is tree-shakable, and option 2 is not.
• Option 2 requires adding another method to Subscriber which isn't generally useful for observable outside of operator creation. And operators aren't really the primary business of observable as a type, even if operators are the primary business of RxJS as a library.

[josepot]

EDIT: After some experimentation and some thought, I'm now leaning towards option 1 above. The reasoning:

I just want to point out that option 1 is 100% compatible and aligned with what it was proposed on #7250 . In fact, if that proposal were to succeed, then the implementation of intercept would be dead simple:

const intercept = <T>(
  prev: Observer<T>,
  next: Partial<Omit<Observer<T>, 'owner'>>
): Partial<Observer<T>> => ({
  ...next,
  owner: prev.owner,
})

[benlesh]

Sorry, this can't happen for reasons I've explained here and here

To elaborate on the "it makes RxJS operators proprietary again idea":

It means that OUR operators will be trying to pass an owner to the subscribe call as part of the Observer (which already doesn't make sense, but I'll get back to that). And it then relies on the implementation of that observable to appropriately handle that owner property on the Observer. No one's observable is going to do that. Only RxJS's operators are dealing with an "owner". Therefore our operator implementations will not work with generalized "subscribables" any longer, despite recent work to make that happen. (outside of a few of our operators that are looking for Observable#pipe, but we will fix that.)

Getting back to why Observer having a public owner property it's providing to the subscription during subscribe:

The Observer itself is an actor. It "observes". The only things it should provide to anyone interacting with it is that which is necessary to "tell" it what it's "observing". "This is the next value", or "I'm complete, nothing left to observe", or "I've had an error". The observer takes in information. Contractually, it does NOT provide information. The idea of an owner property on the Observer makes no sense and breaks the role of observer. It confuses observation with subscription configuration.

Therefor, I would never be okay with adding an owner property or the like to Observer. I'm okay with adding things that are related to observation. "I've observed I'm finalized" for example. Possibly "I've observed the subscription started" but that one is even debatable.

At the point that we do need to pass an "owner" or maybe an AbortSignal or the like to the subscriptoin, it would be through a separate configuration object.

Hopefully this is finally a good enough explanation.

[jakovljevic-mladen]

My two cents:

• solution 1;
• use chain (or operate);
• use named parameters, e.g:

export function chain<T>(destination: Subscriber<any>, on: Partial<Observer<T>> & { finalize?: () => void }): Subscriber<T> {
  return new OperatorSubscriber(destination, on.next, on.complete, on.error, on.finalize);
}

[benlesh]

At this point, I've settled on the following:

export function myMap<In, Out>(project: (value: In) => Out): OperatorFunction<In, Out> {
  return (source) => new Observable(destination => {
    source.subscribe(operate({
      destination,
      next: value => destination.next(project(value))
    }));
  })
}

Will land a PR around this soon.