Partition overload that supports creating more than two observables based on index returned

[Laurens-makel]

Currently we can use partition to create two observables based on true/false.

docs:

const observableValues = of(1, 2, 3, 4, 5, 6);
const [evens$, odds$] = partition(observableValues, value => value % 2 === 0);

I would like to introduce a new overload that would return a numeric value, which should be considered the index of return array.

const observableValues = of(1, 2, 3, 4, 5, 6);
const [low$, mid$, high$] = partition(observableValues, value => value <=2 ? 0 : value <= 4  ? 1 : 2);

Would result in

low$ = [1,2]
mid$ = [3,4]
high$ = [5,6]

[demensky]

I do not think that's possible. Overloads are calculated at runtime, and I don't see any way to calculate the length of an array. It might make sense to pass the length as a separate parameter.

const observableValues = of(1, 2, 3, 4, 5, 6);
const [low$, mid$, high$] = partition(observableValues, someFn, 3);

[Laurens-makel]

I do not think that's possible. Overloads are calculated at runtime, and I don't see any way to calculate the length of an array. It might make sense to pass the length as a separate parameter.

const observableValues = of(1, 2, 3, 4, 5, 6);
const [low$, mid$, high$] = partition(observableValues, someFn, 3);

If an overload would not work, what about a new function named partitionByIndex(obs$, fn) or indexPartition(obs$, fn)

[Laurens-makel]

I was inspired by the .reduce() function, where you could create a new array on the fly, but in RxJs context we need to know the amount of observables to create on the forehand.

Reduce version:

  constructor(fileList: FileList){
    const files = Array.from(fileList);

    const [filtered, folders, invalidFiles] = files.reduce((array, file) => {
      const group = this.isFolder(file) ? 1 : this.isInvalid(file) ? 2 : 0;
      Array.isArray(array[group]) ? array[group].push(file) : array[group] = [file];
      return array;
    }, [] as File[][]);
  }

Some experimental RxJs version, I'm not using ObservableInput and innerFrom() here because my IDE was being an idiot.

For now partitions argument default to 3. The partition() method creates two, if you'd need more then use indexPartition(), which defaults to 3 partitions but could allow up to X partitions.

export function indexPartition<T>(
  source: Observable<T>,
  predicate: (value: T, index: number) => number,
  partitions=3
): Observable<T>[] {
  const observables: Observable<T>[] = [];
  const share$ = source.pipe(shareReplay(1));
  for (let i = 0; i < partitions; i++) {
    observables.push(share$.pipe(filter((emission, index) => predicate(emission, index) === i)))
  }
  return observables;
}

  constructor(fileList: FileList){
    const files = Array.from(fileList);

    const [filtered, folders, invalidFiles] = indexPartition(from(files), file => this.isFolder(file) ? 1 : this.isInvalid(file) ? 2 : 0);
    filtered.subscribe(a => console.log("a: " + a));
    folders.subscribe(b => console.log("b: " + b));
    invalidFiles.subscribe(c => console.log("c: " + c));
  }

You could go even further and accept a string[] as partition keys:

export function groupPartition<T>(
  source: ObservableInput<T>,
  predicate: (value: T, index: number) => number|string,
  keys: number|unknown[] = 3
): ObservableInput<T>[] {
  const share$ = from(source).pipe(shareReplay(1));

  const partitions: unknown[] = Array.isArray(keys)
    ? keys
    : [...Array(keys).keys()]

  return partitions.map(partition => share$.pipe(
    filter((emission, index) => predicate(emission, index) === partition)
  ));
}
constructor(fileList: FileList){
    const files = Array.from(fileList);

    const [a$,b$,c$] = groupPartition(
      from(files), 
      file=> this.isFolder(file) ? 1 : this.isInvalid(file) ? 2 : 0
    );
    a$.subscribe(a => console.log("a: " + a));
    b$.subscribe(b => console.log("b: " + b));
    c$.subscribe(c => console.log("c: " + c));

    const [d$,e$,f$] = groupPartition(
      from(files), 
      file=> this.isFolder(file) ? 'folder' : this.isInvalid(file) ? 'invalid' : 'file', 
      ['folder', 'invalid', 'file']
    );
    d$.subscribe(a => console.log("a: " + a));
    e$.subscribe(b => console.log("b: " + b));
    f$.subscribe(c => console.log("c: " + c));
  }

[voliva]

Yes, the problem is that you need to pass in the number of partitions beforehand....

I would suggest using partitionByKey from @react-rxjs/utils: https://react-rxjs.org/docs/api/utils/partitionByKey. This returns a function to get the partition, which has the advantage that you don't need to pass in information beforehand. Your examples become:

    const [getPartition$] = partitionByKey(
      from(files), 
      file=> this.isFolder(file) ? 'folder' : this.isInvalid(file) ? 'invalid' : 'file'
    );
    getPartition$('folder').subscribe(a => console.log("a: " + a));
    getPartition$('invalid').subscribe(b => console.log("b: " + b));
    getPartition$('file').subscribe(c => console.log("c: " + c));

[Laurens-makel]

Allright, this does the job as well.

Why is this specific in the react-rxjs package? Wouldn't it be nice if this was moved to the generic package?

[voliva]

Historical reasons... it's technically in a separate package from react. We might move it to @rx-state namespace - we already have all the non-react core stuff in @rx-state/core and the react bindings in @react-rxjs/core

[benlesh]

groupBy would handle what you're trying to do. Is there a reason you didn't want to use groupBy?

Otherwise, it's just share then filter if you really have to have them in advance. The issue with share being mostly with completely synchronous observables not really multicasting (as they're "complete" and "ref count zero" at the end of each subscription, because they're synchronous). But... don't use completely synchronous observables. That's silly.