FleuveJS

A simple Observable Utility library.

Why?

Observable is called that way because of the philosophy behind: every Observable is a data stream, potentially infinite. You can add your own data to the Observable, and everyone on the Observable will be notified.

You can also add some pipeline to the Observable, so what you receive has been processed in a convenient way before.

You can bind an Observable behavior to the user's behavior.

An Observable can be forked into multiple other Observables: each Observable child will dispatch some data everytime their parent dispatch some data. You can also close your stream of data, and everyone will know the Observable is complete.

In short:

potentially infinite source of data;
cascading Observables by forking a data stream;
data pre-processing;
aims to allow practically anything as a source of data: a function execution, a user's action, a scalar value, a websocket...

Any contribution and / or feedback is welcome!

Installation

npm i fleuvejs

Or, if you'd prefer to work on a vanilla project:

import {of, map, mutable} from 'https://unpkg.com/fleuvejs@latest/bundle/observable.bundle.js';

Observables, MutableObservables and ObservableForks

Observables are objects containing an inner sequence. Their sequence is finite, and they are immutable. MutableObservables are objects containing an inner sequence too, except this one can be mutated over time. It is infinite, and can be completed with the .close() method. ObservableFork are objects created from either Observable, MutableObservable or ObservableFork:

they can be closed;
they can be potentially infinite;
they cannot be mutated;
they can come with pre-processing operations: when the source emits a new value, the pre-processing operations will be executed on it before being passed to the fork's subscribers.

How To Use Observables

Instantiate an Observable

const temperatures$ = of(10, 20, 13, 24);

Pipe the Observable

You can create a new Observable with the pipe method.

const obs$ = of(18729);
const sum$ = obs$.pipe(
    map((x) => (x + '').split('')), 
    map((numbers) => numbers.reduce((acc, curr) => acc + curr, 0))
);

Subscribe to the Observable

const obs$ = of(12);
obs$.subscribe({
    next: (value) => console.log(value), 
    error: (err) => console.error(err), 
    complete: () => console.log('observable complete')
});

const empty$ = of();
empty$.subscribe((value) => console.log(value)); // will never execute;

// This one will throw an error
obs$.subscribe(42);

// You can as well create a Subscriber object
obs$.subscribe(subscriberOf((x) => console.log(x)));

How To Use MutableObservables

MutableObservable simply extends Observable. All of the previous sections also apply to MutableObservable.

Provide new values with `next` and `compile`

const obs$ = mutable(0);
obs$.next(12, 13, 14, 15, 16); // obs$ inner sequence will now be [ 12, 13, 14, 15, 16 ]
obs$.compile(map((x) => x + 1), map((x) => x * 2)); // obs$ inner sequence will now be [ 26, 28, 30, 32, 34 ]

Close a MutableObservable with `close`

const obs$ = mutable(0);
obs$.close();

How to use ObservableForks

`fork` observables

const obs$ = mutable(12);
const forked$ = fork(obs$, filter(x => x > 15));
forked$.subscribe(x => console.log(x)); // nothing would happen at first
obs$.next(20); // now, 20 would be printed in the browser's console

Stop a fork

No more values will be allowed and the forks will be flagged as complete.

const obs$ = mutable(12);
const fork1$ = fork(obs$, map(x => x * 2));
const fork2$ = fork(obs$, filter(x => x > 100));

const subscriber = subscriberOf((x) => console.log('fork1$ value', x), () => console.log('fork1 complete'));
fork1$.subscribe(subscriber); // will display "24"

fork2$.subscribe((x) => console.log('fork2$ value', x)); // will display nothing

obs$.close(); // will trigger fork1$'s complete callback
obs$.next(99); // the forks' subscribers won't be triggered

Operators

`of` - static

This operator is static: it means you cannot use it as a parameter for methods such as pipe, compile or fork.

This operator allows you to create an Observable from discrete values. It creates a finite Observable. Once created, the Observable is automatically complete.

const obs$ = of(12, 13, 14);
obs$.subscribe(subscriberOf((x) => console.log(x))); // will display "12", "13", "14"