Rxjs: partition makes no sense as a lettable/pipeable operator

[cartant]

RxJS version: 5.5.1

partition is one of the new lettable/pipeable operators in the operators directory. However, it's not really lettable, as it does not return an Observable:

export function partition<T>(
  predicate: (value: T, index: number) => boolean,
  thisArg?: any
): UnaryFunction<Observable<T>, [Observable<T>, Observable<T>]> {
  return (source: Observable<T>) => [
    filter(predicate, thisArg)(source),
    filter(not(predicate, thisArg) as any)(source)
  ] as [Observable<T>, Observable<T>];
}

And with it defined as it is, the usage is somewhat clumsy:

const source = of(1, 2, 3, 4);
const [odds, evens] = partition(value => value % 2)(source);

Are all of the operators in the operators directory supposed to be lettable? Should it be defined elsewhere? And in a way that makes its calls more straightforward? That is, is there any point in its returning a function if that function cannot be passed to pipe (or to let)?

The solution for toPromise was to move it to the prototype, but is there are alternative solution for these types of 'operators' (if that term is even appropriate)?

[felixfbecker]

I wonder if it would make sense to change pipe to really be nothing but a ponyfill for the pipeline operator, i.e. don't require that the output is an Observable. That would make it work more like lodash's flow:

const source = of(1, 2, 3, 4)

// partition

const [odds, evens] = flow(
  map(x => x * 2),
  partition
)(source)

const [odds, evens] = source
  .pipe(map(x => x * 2))
  .pipe(partition)

// toPromise

await flow(
  map(x => x * 2),
  toPromise
)(source)

await source
  .pipe(map(x => x * 2))
  .pipe(toPromise)

[cartant]

I would like to change the typings for pipe so that it can return a ConnectableObservable - see issue #2972.

Changing to something like this would do it:

pipe<A, OA extends Observable<A>>(
  op1: UnaryFunction<Observable<T>, OA>
): OA;
pipe<A, B, OB extends Observable<B>>(
  op1: OperatorFunction<T, A>,
  op2: UnaryFunction<Observable<A>, OB>
): OB;
// etc.

It should be straight forward to provide some additional overload signatures so that the last operator can return a something other than an observable - which could then be returned by pipe.

Whether or not this is desirable, I don't know. I've not spent too much time weighing up the pros and cons. However, if toPromise were to be usable with pipe - rather than on the prototype - it could be used with non-RxJS observables that could be effected from the pipeline, but whether or not that's something that's likely, I don't know.

Given that I cannot recall using toPromise outside of Mocha integration tests, I would take some small delight from seeing it removed from the prototype, but it's not something I care deeply about.

[MorleyDev]

As a thought, the suggestion by @felixfbecker to make pipe a |> ponyfill would make the Observable.pipe function behave in a way consistent with the existing pipe function found in rxjs/util/pipe, which can be used in this manner.

[ivan7237d]

If the pipe signature looks like this, it should also work well with Observable subclasses ConnectableObservable and GroupedObservable:

pipe(): this;
pipe<A>(a: UnaryFunction<this, A>): A;
pipe<A, B>(a: UnaryFunction<this, A>, b: UnaryFunction<A, B>): B;
...

[maxime1992]

Indeed, I thought I could do something like that:

of(1, 2, 3, 4)
  .pipe(
    partition((x) => x % 2 === 0)
  );

But it ends up with the following error:

Argument of type 'UnaryFunction, [Observable, Observable]>' is not assignable to parameter of type 'UnaryFunction, Observable>>'.
Type '[Observable, Observable]' is not assignable to type 'Observable>'.
Property '_isScalar' is missing in type '[Observable, Observable]'.

Repro: https://stackblitz.com/edit/angular-fwfubg?file=app%2Fapp.component.ts

[crimx]

Same here. Need more clarification.

[bratter]

@maxime1992, @crimx, not sure if you were suggesting you'd solved or not, but if not... you need to do as follows:

const source = of(1, 2, 3, 4);
const [odds, evens] = partition(val => val % 2)(source);

Note that the partition function returns a function that has to then be called with the source observable as per felix's comment above, and therefore cannot be used with pipe().

[elpddev]

Using @bratter defintion of the partition function, if this is useful for anyone, for now I am using the partition function in the pipe using a custom function that catch the incoming observable and returns an observable of the partition result.

source$.pipe(
  (obs$) => of(partition(x => isCondition(x))(abs$)),
  mergeMap(([onTrue$, onFalse$]) => merge(
    onTrue$.pipe(tap(x => console.log('on true', x))),
    onFalse$.pipe(tap(x => console.log('on false', x))),
  )),
);

[benlesh]

Current partition should probably be static.

If it's an operator, it should be (source: Observable<T>) => Observable<[Observable<T>, Observable<T>]>, no? Like a specialized groupBy.

[cartant]

@benlesh Static would make the most sense, I think. Essentially, it's the reverse of merge.

[claudiordgz]

@cartant @benlesh When you say static do you mean a function in src/internal/observable?

I was able to use partition as follows:

const source = from([1, 2, 3, 4, 5, 6])
const [evens, odds] = partition((val: any) => val % 2 === 0)(source)

https://stackblitz.com/edit/partition?file=index.ts

If it was a function in src/internal/observable akin to iif would it need to have a scheduler?

[benlesh]

@claudiordgz I think we're talking about partition(source, predicate)

[claudiordgz]

@benlesh thank you, last question... would it be under observable or under util?

[cartant]

Closing this, as partition is to be deprecated. See #3797 and #3807.