How to use the takeUntil function from rxjs

limiter.js

GitHub: openforis/sepal

41
42
43
44
45
46
47
48
49
50
return {
    limiterService,
    limiter$: (observable$, id = uuid(), username) => {
        const stop$ = new Subject()
        return service.submit$(limiterService, {id, username: username || 'ANON'}).pipe(
            takeUntil(stop$),
            mergeMap(() => observable$.pipe(
                finalize(() => stop$.next())
            ))
        )

45

178

34

pool.js

GitHub: openforis/sepal

30
31
32
33
34
35
36
37
38
39

unlock$.pipe(
    tap(instance => instance.locked = false),
    mergeMap(instance =>
        timer(maxIdleMilliseconds).pipe(
            takeUntil(lock$.pipe(
                filter(currentInstance => currentInstance === instance)
            )),
            map(() => instance)
        )

45

178

34

taskRunner.js

GitHub: openforis/sepal

76
77
78
79
80
81
82
83
84
85
const finalState$ = new BehaviorSubject(completedState)

return concat(
    of(initialState),
    progressState$.pipe(
        takeUntil(cancel$.pipe(
            tap(() => finalState$.next(cancelState))
        )),
        catchError(e => concat(finalize$, throwError(() => e)))
    ),

45

178

34

stream.js

GitHub: openforis/sepal

43
44
45
46
47
48
49
50
51

const handleHttp = async ctx => {
    const close$ = new Subject()
    ctx.req.on('close', () => close$.next())
    const body$ = ctx.result$.pipe(
        takeUntil(close$)
    )
    await renderStream(ctx, body$)
}

45

178

34

+ 3 other calls in file