Usage

Wrap

You can Rx.wrap a reactive (or raw source) in an object for fluent, chainable function calls.

After calling wrap, you can invoke operators directly, for example:

// Print out number of clicks in a 200ms time-window
Rx.wrap(Rx.From.event(document, `pointerup`, { x:0, y:0 }))
 .chunk({ elapsed: 200 })
 .transform(v => v.length)
 .onValue(v => { console.log(v) });

Sources can be another wrapped reactive, a normal reactive, iterable, generator, array or function that returns a value.

For example:

Rx.wrap([1,2,3,4,5,6])
  .chunk({ quantity: 2 })
  .onValue(v => {
    // Reads the source array in arrays of length 2
  });

Rx.wrap is nice because you get good code-completion and it can be very concise. The downside is that you don’t get access to the individual parts of a flow if they need to be manipulated after creation.

Run

Compose a source and set of operators together with Rx.run . It returns a new reactive emitting the value of the last operator in the list.

const rx = Rx.run(
  Rx.From.array([ 5, 2, 3, 9, 5]), // Source
  Rx.Ops.max() // One or more operators to follow
);
rx.onValue(v=>{
  // 5, 9
})

Writable

Rx.run will return a reactive that essentially captures the last operator in the list. It doesn’t make sense to write to this reactive since it represents the tail-end.

If you need to be able to write to the head of a composed flow and read the final result, use Rx.writable . Naturally, the source needs to be a writable one.

// Only outputs a value if the maximum changes
const rx = Rx.writable(
  Rx.From.number(10),
  Rx.Ops.max()
);

rx.onValue(v => {
  console.log(v);
});

rx.set(11);   // prints 11
rx.set(5);    // doesn't print anything
rx.set(100);  // prints 100

Fine-grained

Most of the given examples of Rx.run and Rx.writable create and compose sources and operators in one go. But we can use sources and operators more directly if needed:

// Reactive object
const source = Rx.From.object({ size: 10, colour: `red` });

// Transform input objects' 'size' property
const transform = Rx.transform(source, v => ({...v, size: v.size * 2 }))

// The 'transform' stream will emit a value of 28
source.updateField(`size`, 14);