⚡ rx-js-light

rx-js-light is simply the fastest and smallest javascript library for Reactive Programming, providing different tools to generate, consume, and pipe Observables and Observers.

If Reactive Programming does not tell you much or is a new concept to you, you may take a look at this tutorial. In a few words, if you deal frequently with async programming like events, timeouts, promises or streams (ex: front-end development), then rx-js-light is the perfect candidate for you.

Example: emulate double click

const subscribe = pipe$$(fromEventTarget(window, 'click'), [
  bufferTime$$$(500),
  filter$$$((events: PointerEvent[]) => events.length === 2),
  map$$$((events: PointerEvent[]) => events[events.length - 1]),
]);

subscribe((event: PointerEvent) => {
  console.log('double click', event);
});

Click here to see the live demo

Give it a try, and you'll love it !

📖 Table of content

• Introduction

• Installation

• Your first Observable

• Using the built-in Observables

• Emitting values using sources

• Shortcuts

• A practical example for rx-js-light

• Notifications replace RxJS events

• Migrating from rxjs to rx-js-light

• From Promise to rx-js-light

• CHANGELOG

📦 Installation

yarn add @lifaon/rx-js-light
# or
npm install @lifaon/rx-js-light --save

Click here to read the installation manual

📕 Documentation

• Observable
• Observer
• ObservablePipe (ak: Pipeable Operator)
• pipeObservable (ak: Observable.pipe)
• pipeObservablePipes (ak: pipe function)
• Notification (ak: next, complete and error)
• MulticastSource (ak: Subject)
• ReplayLastSource (ak: BehaviorSubject)
• Subscription (kind of: Subscription)

Most of public functions or interfaces have their own documentation into a .md file in their respective directories.

🧲️ Pick the right function

I want to:

create an Observable from

without notifications

• nothing and emit no value: empty

• one value

  • already defined: single 🔥
  • defined later: reference

• a list of values: of 🔥

• an iterable

  • array: fromArray
  • iterable: fromIterable
  • iterator ⚠️: fromIterator

• something related to the DOM

  • an EventTarget: fromEventTarget 🔥
  • an IntersectionObserver: fromIntersectionObserver
  • a ResizeObserver: fromResizeObserver
  • a css @media query: fromMatchMedia

• one Observable

  • defined later: defer

• many Observables. When any value is received

  • re-emit it concurrently: merge 🔥
  • combine the values in an array and emit it: combineLatest 🔥
  • combine the values in an array, runs a function with these values, and emit distinct returned values: reactiveFunction 🔥
    • arithmetic: reactiveAdd, reactiveSubtract, reactiveMultiply, reactiveDivide
    • logic: reactiveAnd, reactiveOr, reactiveNot
    • comparison: reactiveEqual, reactiveNotEqual, reactiveGreaterThan, reactiveGreaterThanOrEqual, reactiveLowerThan, reactiveLowerThanOrEqual
    • string: reactiveTemplateString

• time related

  • emits every 'ms': interval 🔥
  • emits after 'ms': timeout
    • with a complete notification: timeoutWithCompleteNotification
    • with an error notification: timeoutWithErrorNotification
  • emits when idle time is available: idle
  • emits on each animation frame: fromAnimationFrame

with notifications

• nothing and send a complete Notification: emptyWithNotifications 🔥

• one value

  • already defined: singleWithNotifications 🔥

• a list of values: ofWithNotifications 🔥

• an iterable

  • sync

    • array: fromArrayWithNotifications
    • iterable: fromIterableWithNotifications
    • iterator ⚠️: fromIteratorWithNotifications

  • async

    • async iterable: fromAsyncIterable
    • async iterator ⚠️: fromAsyncIterator

• something related to the DOM

  • the position of the user (Geolocation): fromGeolocationPosition

• a promise

  • with a factory: fromPromiseFactory 🔥
  • without a factory ⚠️: fromPromise 🔥

• a readable stream

  • w3c streams
    • readable stream: fromReadableStream
    • readable stream reader ⚠: fromReadableStreamReader
  • nodejs: TODO

• an http request

  • using fetch: fromFetch 🔥
    • fromFetchJSON
    • fromFetchText
    • fromFetchStream
  • using xhr: fromXHR

• a blob (reads content): readBlob

• an error (send an error Notification): throwError 🔥

• many Observables:

  • awaits that all Observables complete and then sends the last received values as an array: forkJoin 🔥
  • awaits that one Observable completes and then sends the last received value: raceWithNotifications 🔥

convert an Observable to

without notifications

• a promise: toPromise

with notifications

• a promise

  • with only the last value: toPromiseLast 🔥
  • with every value: toPromiseAll

• an async iterable: toAsyncIterable

create an ObservablePipe which

without notifications

• ObserverPipe related

  • emits only distinct received values: distinctObservablePipe 🔥
  • filters received values: filterObservablePipe 🔥
  • transforms received values: mapObservablePipe 🔥
  • transforms and filter received values: mapFilterObservablePipe
  • transforms received values with an accumulator: scanObservablePipe
  • reads received values, and re-emits them without transformations: tapObservablePipe

• Source related

  • allows one Observable to emit its values to many Observable: shareObservablePipe 🔥

• reduces the order of an Observable of Observables

  • once: mergeAllSingleObservablePipe 🔥
  • many: mergeAllObservablePipe
  • maps an Observable and then reduces the order:
    • once: mergeMapSingleObservablePipe 🔥
    • many: mergeMapObservablePipe

• accumulates values:

  • until another Observable emits: bufferObservablePipe
  • within a fixed period of time: bufferTimeObservablePipe

• take the first value: firstObservablePipe

• take the X first values: takeObservablePipe

• take the values until another observable emits: takeUntilObservablePipe

• take the first value that passes a condition: findObservablePipe

• time related

  • debounceTimeObservablePipe 🔥
  • throttleTimeObservablePipe 🔥
  • debounceFrameObservablePipe
  • debounceImmediateObservablePipe
  • debounceMicrotaskObservablePipe

• logs the state of the upper Observable: logStateObservablePipe

with notifications

• awaits to receive a complete or error Notification, and performs some kind of mergeMap:
  • thenObservablePipe 🔥
    • fulfilledObservablePipe
    • rejectedObservablePipe
    • finallyObservablePipe

emit a value myself => create a Source which emits values to

• multiple Observers: createMulticastSource
• only one Observer: createUnicastSource
• one or many Observer and stores all emitted values: createReplaySource, createMulticastReplaySource, createUnicastReplaySource
• one or many Observer and stores the last emitted value: createReplayLastSource, createMulticastReplayLastSource 🔥, createUnicastReplayLastSource

create an io stream

• from a websocket: createWebSocketByteStream

others

• chain many ObservablePipes: pipeObservablePipes
• chain many ObservablePipes with an Observable: pipeObservable 🔥

🔥: most important functions

Can't find a function that suits your needs ? Open a discussion, or create your own and share it !

Differences with RxJS:

• no classes: this choice allows blazing fast performances and very small bundle size. Indeed, creating a class with the new keyword is slow, and method names can't be mangled (minimized), where function calls are really well optimized by javascript engines. However, it has a minor cost: chaining operators or method calls are done through functions, which is a little less elegant (in terms of code readability).

• no next, complete and error: instead this lib uses notifications. In reality, not all Observables require to emit a final state. For example, the RxJS interval never reaches a complete state. It just sends numbers. Moreover, some Observables may want to emit more than this 3 events: we may imagine an XHR Observable which emits an upload-progress and download-progress events.

• some concepts / operators / methods may have a different behaviour or name. Take care to read the documentation before any hasty use.