Supporting the TC39 |> pipeline operator

[benlesh]

Goals

• Align with the pipeline proposal
• Provide good developer ergonomics
• Don’t impact performance or readability anymore than is necessary
• Limit breaking changes.

The current pipeline proposal isn’t compatible with “pipelining” existing pipeable functions

The current TC39 proposal for pipelines is at stage 2, and unfortunately it doesn’t support “pipeable functions”. It looks something like this if used with our current operators:

(source$
  |> map(x => x * 2)(%)
  |> filter(x => x > 10)(%)
).subscribe(console.log)

Where % (or maybe some other character like ^) conveys where the result of the left hand side is used in the right hand side.

If we support this, it might be pretty nice that people can do quick and easy calls like concatMap(source, fn) and that concat and merge would “just work” as operators as well.

Proposal 1: Automatically curry if called with fewer values

1. We can remake our operators in a non-higher-order format, where the first argument is the source observable.
2. We wrap those operators in a function that makes it so if they’re called without the first argument (basically one less argument than expected), it returns a unary function that accepts the last argument.

So in this world, something like concatMap would look like this:

// Note the signature change.
declare const concatMap: <T, R>(source: Observable<T>, project: (value: T, index: number) => ObservableInput<R>) => Observable<R>

concatMap(source$, v => of(v, v, v))
source$.pipe(concatMap(v => of(v, v, v))
concatMap(v => of(v, v, v))(source$) // still works.

Issues

• TypeScript typings might get weird.
• Possible performance impact when initially creating the observable with operators.
• Operators that accept only rest arguments will not work, they would need to be refactored to take an array of arguments. We can get around this if the rest arguments have at least one required argument — which they likely do — basically fn(a: any, rest: …args: any[]), but this isn’t ergonomic because we’ll have to train people how to properly make operators for use here. It’s quirky.

Impl Notes

We could use a simple function like this to convert what are basically “observable creation” functions — where the source is the first argument — into functions that if they’re called with one fewer argument than is expected returns a function that accepts the first argument:

function maybeCurry(fn: (…args: any[]) => any) {
  const expectedArgs = fn.length

  return (…args: any[]) => {
    if (args.length === expectedArgs - 1) {
      return (first: any) => fn(first, …args)
    }
    return fn(…args)
  }
}

Proposal 2: Export a “standalone” and “pipeable” version of every operator

In this proposal, there would be one “standalone” implementation of everything, but we’d export a curried version of it alongside.

Basically:

function fromConcatMap<T>(source: Observable<T>, project: (value: T, input: number) => ObservableInput<R>): Observable<R> {
  // imple
}

const concatMap = makePipeable(fromConcatMap)

fromConcatMap(source$, v => of(v, v, v))
source$.pipe(concatMap(v => of(v, v, v))
concatMap(v => of(v, v, v))(source$) // still works.

Issues

• We basically DOUBLE the already huge surface area of RxJS overnight.
• We’d need to dream up a naming convention that isn’t awful — I don’t like the fromConcatMap above.

Proposal 3: Ignore |> until it lands, then pivot

Here we just wait until |> lands in most environments, then pivot our library to work with pipelining — and only pipelining — in a major version as a breaking change.

Proposal 4: We ignore |> and stick with what we’re doing

In this scenario, we just accept that the TC39 has failed to deliver a useful feature and we continue doing what we’ve been doing until hopefully something else lands that helps us.

[demensky]

Thanks to the pipeline operator, it is possible to erase the line between "function" and "operator".

There will be no need for operators like zipWith, combineLatestWith, etc.

source$.pipe(zipWith(foo$, bar$));

// turn into

source$ |> zip(%, foo$, bar$);

It also allows you to choose the order of the arguments yourself.

sample(source$, notifier$);
source$ |> sample(%, notifier$);
notifier$ |> sample(source$, %);

I also don't like the naming idea (fromConcatMap).
I suggest make separate export like rxjs/pipeline.

[benlesh]

CORE TEAM: Proposal 3 or 4 for now. Do nothing.

[czabaj]

I encountered this problem when working with RxJS in Rescript, Rescript features a pipe operator, but it uses the data first approach, i.e. you must pipe to a function and the LHS of the pipe (data) is passed as first argument to the RHS of the pipe (function), so piping into operators does not work, as these have typically a signature

operator(fn)(xs)

where with the data-first pipe

xs->operator(fn)

leads to

operator(xs, fn)

which is wrong because it does not apply xs partially and it flips the arguments position. You can go with argument placeholder (like % in the TC39 proposal) but that leads to more verbose code, so I ended with a helper function, which I named op, which applies the arguments in proper order and applies them partially, it basically does just

let op = (xs, operator) => operator(xs)

and the usage

xs->op(operator1)->op(operator2)

it works well, it is short and the Rescript compiler is in some cases capable of unravel the op from the produced JavaScript, so the cost of a function call is eliminated.

I think this would also work well for the {Java,Type}Script.

[x2764tech]

Am I missing something?

At lot of the examples you've shared boil down to:

of(1,2,3) 
  |> %.pipe(map(x => x * 2)) 
  |> %.pipe(map(value => console.log(value)))

Hence conversion from an rxjs operator to a TS pipable function is (very roughly)

const toTypescriptPipable = <In,Out>(operator: OperatorFunction<In, Out>) 
  => (source: Observable<In>) 
  => source.pipe(operator)

However, it would be worth waiting until there's implementation details before considering the best solution - if it's implemented like Iterables, then it might be that Observables can override a [Symbol.Pipe](operator) function, in which case it's just delegating to pipe.