How to use the switchMap function from rxjs

task.js

GitHub: openforis/sepal

43
44
45
46
47
48
49
50
51
52
    })

const monitor$ = taskId =>
    interval(MONITORING_FREQUENCY).pipe(
        exhaustMap(() => status$(taskId)),
        switchMap(({state, error_message: error}) =>
            (error || state === FAILED)
                ? throwError(() => new Error(error))
                : of(state)
        ),

45

178

34

+ 7 other calls in file

cloudStorageDownload.js

GitHub: openforis/sepal

18
19
20
21
22
23
24
25
26
27
    )
})
const download$ = ({bucketPath, prefix, downloadDir, deleteAfterDownload}) =>
    cloudStorage$().pipe(
        map(cloudStorage => cloudStorage.bucket(`gs://${bucketPath}`)),
        switchMap(bucket =>
            do$(
                `download: ${JSON.stringify({bucketPath, prefix, downloadDir, deleteAfterDownload})}`,
                bucket.getFiles({prefix, autoPaginate: true})
            ).pipe(

45

178

34

+ 9 other calls in file

pool.js

GitHub: openforis/sepal

107
108
109
110
111
112
113
114
115
116
}

return {
    getInstance$: () =>
        getInstance$().pipe(
            switchMap(instance =>
                concat(of(instance), NEVER).pipe(
                    tap(instance => lock(instance)),
                    map(({item}) => item),
                    finalize(() => release(instance))

45

178

34

userStore.js

GitHub: openforis/sepal

58
59
60
61
62
63
64
65
66
67
firstValueFrom(
    get$(currentUserUrl, {
        headers: {[SEPAL_USER_HEADER]: JSON.stringify(user)}
    }).pipe(
        map((({body}) => JSON.parse(body))),
        switchMap(user => {
            log.isDebug() && log.debug(`${usernameTag(user.username)} ${urlTag(req.url)} Updated user in user store, connected to Google: ${!!user.googleTokens}`)
            return from(setUser(user))
        }),
        catchError(error => {

45

178

34

auth.js

GitHub: openforis/sepal

25
26
27
28
29
30
31
32
33
34
return postJson$(refreshGoogleTokensUrl, {
    headers: {
        [SEPAL_USER_HEADER]: JSON.stringify(user)
    }
}).pipe(
    switchMap(({body: googleTokens}) => {
        if (googleTokens) {
            log.debug(() => `${usernameTag(user.username)} ${urlTag(req.originalUrl)} Refreshed Google tokens`)
            const updatedUser = {...user, googleTokens: JSON.parse(googleTokens)}
            res.set(SEPAL_USER_HEADER, JSON.stringify(updatedUser))

45

178

34

+ 3 other calls in file

fileSystem.test.js

GitHub: openforis/sepal

13
14
15
16
17
18
19
20
21
22
)

const basename = 'some_dir'
stream('emits basenames when dir is non-empty',
    () => mkdir$(Path.join(emptyDirPath, basename)).pipe(
        switchMap(() => ls$(emptyDirPath))
    ),
    emitsOne(files => {
        expect(files).toEqual([basename])
    })

45

178

34

+ 11 other calls in file

ws.js

GitHub: openforis/sepal

53
54
55
56
57
58
59
60
61
62
    timer(0, pollIntervalMilliseconds).pipe(
        takeWhile(() => state.enabled)
    )

trigger$.pipe(
    switchMap(trigger =>
        concat(triggerAction$(trigger), poll$)
    ),
    exhaustMap(() => from(scanDirs())),
    distinctUntilChanged(_.isEqual),

45

177

0

+ 3 other calls in file

changeAlerts.js

GitHub: openforis/sepal

25
26
27
28
29
30
31
32
33
34
        map(image => image.retile(32)),
    )
},
getBands$() {
    return delegate$.pipe(
        switchMap(delegate => delegate.getBands$())
    )
},
getVisParams$() {
    return delegate$.pipe(

45

177

0

+ 5 other calls in file

taskManager.js

GitHub: openforis/sepal

100
101
102
103
104
105
106
107
108
109
    )
).pipe(
    shareReplay()
)
return executeTask$({task, cmd$: taskCancellation$}).pipe(
    switchMap(progress =>
        progress.state === 'COMPLETED'
            ? taskCompleted$(task.id)
            : progress.state === 'CANCELED'
                ? taskCanceled$(task.id)

45

177

0

+ 2 other calls in file

drive.js

GitHub: VEGMQAZ/sepal

31
32
33
34
35
36
37
38
39
40
    return operation$
}

const auth$ = () =>
    defer(getCurrentContext$).pipe(
        switchMap(({userCredentials}) => {
            const oAuth2Client = new google.auth.OAuth2()
            const expiration = moment(userCredentials['access_token_expiry_date'])
            log.debug(() => `Authenticating with token expiring ${expiration.fromNow()} (${expiration})`)
            oAuth2Client.setCredentials({

45

0

0

+ 5 other calls in file

apps.js

GitHub: VEGMQAZ/sepal

42
43
44
45
46
47
48
49
50
51

const updateApp$ = app => {
    log.info('Updating app', app.path)
    return exists$(app).pipe(
        switchMap(cloned => cloned ? of(true) : clone$(app)),
        switchMap(() => checkout$(app)),
        switchMap(() => pull$(app)),
        switchMap(() => useCustomKernel$(app)),
        switchMap(useCustomKernel => useCustomKernel ? updateKernel$(app) : of(false)),
        catchError(error => {

45

0

0

+ 21 other calls in file