Consuming

We’ve already seen the common way to consume value, using rx.onValue and giving a callback function:

rx.onValue(value => {
  // value emitted by reactive 'rx'
})

…but there are other approaches which are sometimes more useful.

Read a single value

Rx.takeNextValue reads a single value from a reactive. You can specify a maximum wait period to avoid freezes.

// Read the next value from 'source', waiting up to 1s
const value = await Rx.takeNextValue(source, 1000);

An error is thrown if the waiting time is exceeded or the reactive closes.

To an array

Rx.toArray reads some values from a source reactive, returning them as an array. Limits can be set for how many items to read or for how long.

// Read 5 items from `source`
const data = await Rx.toArray({ limit: 5 })(source);

Rx.toArrayOrThrow is similar, but will throw an error if the desired number of items is not reached.

To a generator

Rx.toGenerator will wrap the reactive as a generator. This allows values to be iterated over using for of loops, like Chains

// Reactive numerical value
const number = Rx.number(10);

// Get a generator-form:
const g = Rx.toGenerator(number);
for await (const v of g) {
 console.log(v); // Prints out whenever the reactive value changes
}

// Careful! Execution doesn't continue until Reactive finishes