How to use the pipe function from rxjs

messageQueue.js

GitHub: openforis/sepal

2
3
4
5
6
7
8
9
10
11
const {defer, pipe, timer, Subject, firstValueFrom, retryWhen, mergeMap} = require('rxjs')

const EXCHANGE = 'sepal.topic'
const RETRY_DELAY_MS = 10000

const retry = delay => pipe(
    retryWhen(error$ =>
        error$.pipe(
            mergeMap(
                error => {

45

178

34

exhaustMap.md?plain=1

GitHub: observable-playground/observable-playground

31
32
33
34
35
36
37
38
39
exhaustMap$.subscribe(rxObserver('exhaustMap( timer(0, 3).take(3) )'));


// helpers
function colorize(color) {
  return pipe(
    map(y => Marble(y, color))
  );
}

10

159

3

publicTransport.js

GitHub: PredictiveMovement/digital-twin

11
12
13
14
15
16
17
18
19
20
  mergeAll,
} = require('rxjs/operators')
const { error } = require('../lib/log.js')

const reduceMap = (idProp = 'id') =>
  pipe(reduce((map, item) => map.set(item[idProp], item), new Map()))

const addProp = (prop, fn) =>
  pipe(
    map((item) =>

1

16

0

+ 3 other calls in file

index.js

GitHub: lbovet/makine

31
32
33
34
35
36
37
38
39
40
  app[method.toLowerCase()](url, (request) =>
    subject.next(request));
  return subject.pipe(...pipeline, repeat());
},
extract: {
  body: prop => prop ? pipe(pluck('body'), pluck(prop)) : pluck('body'),
  url: () => pluck('url'),
  params: param => param ? pipe(pluck('params'), pluck(param)) : pluck('params')
},
request: {

0

2

3

+ 3 other calls in file

index.js

GitHub: caosbad/api

31
32
33
34
35
36
37
38
39
40
exports.Scheduler = rxjs.Scheduler;
exports.Subscription = rxjs.Subscription;
exports.Subscriber = rxjs.Subscriber;
exports.Notification = rxjs.Notification;
exports.NotificationKind = rxjs.NotificationKind;
exports.pipe = rxjs.pipe;
exports.noop = rxjs.noop;
exports.identity = rxjs.identity;
exports.isObservable = rxjs.isObservable;
exports.ArgumentOutOfRangeError = rxjs.ArgumentOutOfRangeError;

299

0

1