Showcasing DistinctElements and SwitchByMap RX operators

switchByMap.ts

import { red, yellow } from "chalk";
import { differenceWith, isEqual } from "lodash";
import { Observable, of, Subject } from "rxjs";
import {
  concatMap, delay, groupBy, map, mergeMap, scan, switchMap,
} from "rxjs/operators";


/*
  Transform stream of arrays into stream of distinct elements
*/
const distinctElements = () =>
  (source$: Observable<any[]>) =>
  source$.pipe(
    scan<any[], {
      previous: any[];
      difference: any[];
    }>(
      (accumulator, current) => ({
        previous: current,
        difference: differenceWith(current, accumulator.previous, isEqual),
      })
    , {
      previous: [],
      difference: [],
    }),
    map(x => x.difference),
    concatMap(x => x)
  );

/*
  Map to observable, complete previous inner observable within group
*/
const switchByMap = (groupSelector, mapOperation) => source$ => source$.pipe(
  groupBy(groupSelector),
  mergeMap((group$: Observable<User>) => group$.pipe(
    switchMap(mapOperation)
  ))
);


/*
 ================================================================================
*/


type User = {
  name: string;
  status: string;
}

const users$ = new Subject<User[]>();
const status$ = users$.pipe(
  distinctElements()
);
const expensiveOperation = (x) => of(x).pipe(delay(500));


users$.subscribe(
  users => console.log(red("•USERS•"), users)
);
status$.subscribe(
  r => console.log(yellow("•STATUS•"), r)
);
status$.pipe(
  switchByMap(
    user => user.name,
    expensiveOperation
  )
).subscribe(
  r => console.log("•OPERATION•", r)
);


/*
 ================================================================================
*/


users$.next([{
  name: "Ana",
  status: "Running",
}, {
  name: "Tom",
  status: "Swimming",
}, {
  name: "John",
  status: "Painting",
}]);

users$.next([{
  name: "Ana",
  status: "Running",
}, {
  name: "Tom",
  status: "Swimming",
}, {
  name: "John",
  status: "Painting",
}]);

users$.next([{
  name: "Ana",
  status: "Running",
}, {
  name: "Tom",
  status: "Swimming",
}, {
  name: "John",
  status: "Painting",
}, {
  name: "Jack",
  status: "Woke up",
}]);

users$.next([{
  name: "Ana",
  status: "Gardening",
}, {
  name: "Tom",
  status: "Gardening",
}, {
  name: "John",
  status: "Painting",
}, {
  name: "Jack",
  status: "Gardening",
}]);

setTimeout(() => {
  users$.next([{
    name: "Ana",
    status: "Reading",
  }, {
    name: "Tom",
    status: "Gardening",
  }, {
    name: "John",
    status: "Painting",
  }, {
    name: "Jack",
    status: "Gardening",
  }]);
}, 1200);

Output:

•USERS• []
•USERS• [
  { name: 'Ana', status: 'Running' },
  { name: 'Tom', status: 'Swimming' },
  { name: 'John', status: 'Painting' }
]
•STATUS• { name: 'Ana', status: 'Running' }
•STATUS• { name: 'Tom', status: 'Swimming' }
•STATUS• { name: 'John', status: 'Painting' }
•USERS• [
  { name: 'Ana', status: 'Running' },
  { name: 'Tom', status: 'Swimming' },
  { name: 'John', status: 'Cleaning' }
]
•STATUS• { name: 'John', status: 'Cleaning' }
•USERS• [
  { name: 'Ana', status: 'Running' },
  { name: 'Tom', status: 'Swimming' },
  { name: 'John', status: 'Cleaning' },
  { name: 'Jack', status: 'Woke up' }
]
•STATUS• { name: 'Jack', status: 'Woke up' }
•USERS• [
  { name: 'Ana', status: 'Gardening' },
  { name: 'Tom', status: 'Gardening' },
  { name: 'John', status: 'Cleaning' },
  { name: 'Jack', status: 'Gardening' }
]
•STATUS• { name: 'Ana', status: 'Gardening' }
•STATUS• { name: 'Tom', status: 'Gardening' }
•STATUS• { name: 'Jack', status: 'Gardening' }
•OPERATION• { name: 'John', status: 'Cleaning' }
•OPERATION• { name: 'Ana', status: 'Gardening' }
•OPERATION• { name: 'Tom', status: 'Gardening' }
•OPERATION• { name: 'Jack', status: 'Gardening' }
•USERS• [
  { name: 'Ana', status: 'Reading' },
  { name: 'Tom', status: 'Gardening' },
  { name: 'John', status: 'Reading' },
  { name: 'Jack', status: 'Gardening' }
]
•STATUS• { name: 'Ana', status: 'Reading' }
•STATUS• { name: 'John', status: 'Reading' }
•OPERATION• { name: 'Ana', status: 'Reading' }
•OPERATION• { name: 'John', status: 'Reading' }