How to use the firstValueFrom function from rxjs

http.js

GitHub: openforis/sepal

2
3
4
5
6
7
8
9
10
11
const {firstValueFrom} = require('rxjs')
const log = require('#sepal/log').getLogger('http')

const getEmailNotificationsEnabled = async emailAddress => {
    log.debug(() => `Getting email notifications preference for address <${emailAddress}> from origin`)
    const response = await firstValueFrom(
        get$(`https://${sepalHost}/api/user/email-notifications-enabled/${emailAddress}`, {
            username: sepalUsername,
            password: sepalPassword
        })

45

178

34

userStore.js

GitHub: openforis/sepal

53
54
55
56
57
58
59
60
61
62

const updateUser = req => {
    const user = getRequestUser(req)
    if (user) {
        log.isTrace() && log.trace(`${usernameTag(user.username)} ${urlTag(req.url)} Updating user in user store`)
        firstValueFrom(
            get$(currentUserUrl, {
                headers: {[SEPAL_USER_HEADER]: JSON.stringify(user)}
            }).pipe(
                map((({body}) => JSON.parse(body))),

45

178

34

stream.js

GitHub: openforis/sepal

33
34
35
36
37
38
39
40
41
42
const handleSuccess = (ctx, value) =>
    ctx.body = value

const renderStream = async (ctx, body$) => {
    try {
        handleSuccess(ctx, await firstValueFrom(body$, {defaultValue: undefined}))
    } catch (error) {
        handleError(ctx, error)
    }
}

45

178

34

auth.js

GitHub: openforis/sepal

126
127
128
129
130
131
132
133
134
135
    ? verifyGoogleTokens$
    : hasAuthHeaders()
        ? authenticate$
        : send401$

const shouldContinue = await firstValueFrom(
    result$.pipe(
        catchError(error => {
            log.error(`${urlTag(req.originalUrl)} Got an unexpected error when trying to authenticate`, error)
            res.status(500)

45

178

34

messageQueue.js

GitHub: openforis/sepal

85
86
87
88
89
90
91
92
93
94
        }
    })
}

const initMessageQueue = async (amqpUri, {publishers = [], subscribers = []}) => {
    const connection = await firstValueFrom(messageQueue$(amqpUri))
    for await (const {key, publish$} of publishers) {
        await topicPublisher({amqpUri, connection, key, publish$})
    }
    for await (const {queue, topic, handler} of subscribers) {

45

178

34

publicTransport.js

GitHub: PredictiveMovement/digital-twin

32
33
34
35
36
37
38
39
40
41
  serviceDates,
  routeNames,
  excludedLineNumbers,
} = gtfs(operator)

const allTrips = await firstValueFrom(trips.pipe(reduceMap()))
const allRouteNames = await firstValueFrom(routeNames.pipe(reduceMap()))
const allStops = await firstValueFrom(stops.pipe(reduceMap()))
const allServices = await firstValueFrom(serviceDates.pipe(reduceMap('date')))
const todaysServices = allServices.get(date).services

1

16

0

+ 7 other calls in file

symbol-poll.js

GitHub: ymuichiro/symbol_web

4
5
6
7
8
9
10
11
12
13

const repo = new symbol.RepositoryFactoryHttp(process.env.SYMBOL_NODE);

async function getNetworkProperties() {
  const networkRepo = repo.createNetworkRepository();
  const n = await firstValueFrom(networkRepo.getNetworkProperties());
  return {
    identifier: n.network.identifier,
    epochAdjustment: n.network.epochAdjustment,
    generationHashSeed: n.network.generationHashSeed,

5

5

3

+ 5 other calls in file

auto-reward.js

GitHub: ymuichiro/symbol_web

48
49
50
51
52
53
54
55
56
  signedAggregateTx,
  nt
).setMaxFee(100)

const signedLockTx = bot.sign(hashLockTx, ng);
await op.firstValueFrom(transactionHttp.announce(signedLockTx));
console.log(signedLockTx.hash);

const listener = repositoryFactory.createListener();

5

5

3

+ 31 other calls in file

virtualTime.js

GitHub: PredictiveMovement/digital-twin

48
49
50
51
52
53
54
55
56
57
    distinctUntilChanged()
  )
}

getTimeInMillisecondsAsPromise() {
  return firstValueFrom(this.getTimeInMilliseconds())
}

play() {
  this.internalTimeScale = 1

1

16

0

+ 3 other calls in file

fleet.js

GitHub: PredictiveMovement/digital-twin

103
104
105
106
107
108
109
110
111
112
    dispatch(this.cars, this.unhandledBookings)
  ).pipe(share())
}

async canHandleBooking(booking) {
  return firstValueFrom(
    this.cars.pipe(
      first((car) => car.canHandleBooking(booking), false /* defaultValue */)
    )
  )

1

16

0