rx_omit_duplicates.ts

rx_omit_duplicates.ts

import Rx = require('rxjs/Rx');

interface Obj {
  id: number;
}

type Chunk = Obj[]

const chunks$: Rx.Observable<Chunk> = Rx.Observable.from([
  [ { id: 1 }, { id: 2} ],
  [ { id: 3 }, { id: 2} ],
  [ { id: 1 } ],
]);

interface Acc {
  chunk?: Chunk;
  ids: number[];
}

const seed: Acc = {
  ids: [],
}

function omitDuplicates(chunks$: Rx.Observable<Chunk>): Rx.Observable<Chunk> {
  return chunks$
    .scan((acc, chunk) => {
      const filteredChunk = chunk.filter(obj => {
        return acc.ids.indexOf(obj.id) === -1;
      });
      const ids = filteredChunk.map(obj => obj.id);

      return {
        chunk: filteredChunk,
        ids: [...acc.ids, ...ids],
      };
    }, seed)
    .map(acc => acc.chunk);
}

chunks$
  .let(omitDuplicates)
  .subscribe(console.log);