polyrhythm 🎵🎶

polyrhythm is a way to avoid async race conditions, particularly those that arise when building UIs, in JavaScript. It can replace Redux middleware like Redux Saga, and is a framework-free library that supercharges your timing and resource-management. And it's under 4Kb.

Its API is a synthesis of ideas from:

• 💜RxJS. Older than Promises, nearly as old as JQuery.
• 💜Redux-Observable, Redux Saga, Redux Thunk. Async.
• 💙Macromedia Flash. Mutliple timelines.
• 💙JQuery. #on and #trigger.

For use in a React context, polyrhythm-react exports all in this library, plus React hooks for interfacing with it.

Installation

npm install polyrhythm

What Is It?

polyrhythm is a TypeScript library that is a framework-independent coordinator of multiple streams of async using RxJS Observables.

The goal of polyrhythm is to be a centralized control of timing for sync or async operations in your app.

Because of it's pub-sub/event-bus design, your app remains inherently scalable because originators of events don't know about consumers, or vice versa. If a single subscriber errs out, neither the publisher nor other subscribers are affected. Your UI layer remains simple— its only job is to trigger/originate events. All the logic remains separated from the UI layer by the event bus. Testing of most of your app's effects can be done without going through your UI layer.

polyrhythm envisions a set of primitives can compose into beautiful Web Apps and User Experiences more simply and flexibly than current JavaScript tools allow for. All thie with a tiny bundle size, and an API that is delightfully simple.

Where Can I Use It?

The framework-independent primitives of polyrhythm can be used anywhere. It adds only 3Kb to your bundle, so it's worth a try. It is test-covered, provides types, is production-tested and performance-tested.

Declare Your Timing, Don't Code It

RxJS was written in 2010 to address the growing need for async management code in the world of AJAX. Yet in 2021, it can still be a large impact to the codebase to add async to a function declaration, or turn a function into a generator with function*() {}. That impact can 'hard-code' in unadaptable behaviors or latency. And relying on framework features (like the timing difference between useEffect and useLayoutEffect) can make code vulnerable to framework changes, and make it harder to test.

polyrhythm gives you 5 concurrency modes you can plug in trivially as configuration parameters, to get the full power of RxJS elegantly.

The listener option mode allows you to control the concurrency behavior of a listener declaratively, and is important for making polyrhythm so adaptable to desired timing outcomes. For an autocomplete or session timeout, the replace mode is appropriate. For other use cases, serial, parallel or ignore may be appropriate.

If async effects like AJAX were represented as sounds, this diagram shows how they might overlap/queue/cancel each other.

Being able to plug in a strategy ensures that the exact syntax of your code, and your timing information, are decoupled - the one is not expressed in terms of the other. This lets you write fewer lines, be more direct and declarative, and generally cut down on race conditions.

Not only do these 5 modes handle not only what you'd want to do with RxJS, but they handle anything your users would expect code to do when async process overlap! You have the ease to change behavior to satisfy your pickiest users, without rewriting code - you only have to update your tests to match!

Now let's dig into some examples.

Example 1: Auto-Complete Input (replace mode)

Based on the original example at LearnRxjs.io...

Set up an event handler to trigger search/start events from an onChange:

<input onChange={e => trigger('search/start', e.target.value)} />

Listen for the search/results event and update component or global state:

filter('search/results', ({ payload: results }) => {
  setResults(results);
});

Respond to search/start events with an Observable, or Promise of the ajax request. <<<<<<< HEAD

Assign the output to the search/results event, and specify your mode, and you're done and race-condition-free!

on(
  'search/start',
  ({ payload }) => {
    return fetch(URL + payload).then(res => res.json());
  },
  {
    mode: 'replace',
    trigger: { next: 'search/results' },
  }
);

mode:replace does what switchMap does, but with readability foremost, and without requiring you to model your app as a chained Observable, or manage Subscription objects or call .subscribe() or .unsubscribe() explicitly.

Debounced Search CodeSandbox

Example 2: Ajax Cat Fetcher (multi-mode)

Based on an XState Example showing the value of separating out effects from components, and how to be React Concurrent Mode (Suspense-Mode) safe, in XState or Polyrhythm.

Try it out - play with it! Is the correct behavior to use serial mode to allow you to queue up cat fetches, or ignore to disable new cats while one is loading, as XState does? You choose! I find having these options easily pluggble enables the correct UX to be discovered through play, and tweaked with minimal effort.

Cat Fetcher AJAX CodeSandbox

Example 3: Redux Toolkit Counter (multi-mode)

All 5 modes can be tried in the polyrhythm version of the Redux Counter Example Sandbox

Can I use Promises instead of Observables?

Recall the auto-complete example, in which you could create a new search/results event from either a Promise or Observable:

on('search/start', ({ payload }) => {
  // return Observable
  return ajax.get(URL + payload).pipe(
    tap({ results } => results)
  );
  // OR Promise
  return fetch(URL + payload).then(res => res.json())
}, {
 mode: 'replace',
 trigger: { next: 'search/results' }
});

With either the Promise, or Observable, the mode: replace guarantees your autocomplete never has the race-condition where an old result populates after new letters invalidate it. But with an Observable:

• The AJAX can be canceled, freeing up bandwidth as well
• The AJAX can be set to be canceled implicitly upon component unmount, channel reset, or by another event declaratively with takeUntil. And no Abort Controllers or await ever required!

You have to return an Observable to get cancelation, and you only get all the overlap strategies and lean performance when you can cancel. So best practice is to use them - but they are not required.

UI Layer Bindings

trigger, filter listen (aka on), and query are methods bound to an instance of a Channel. For convenience, and in many examples, these bound methods may be imported and used directly

import { trigger, on } from 'polyrhythm';
on(...)
trigger(...)

These top-level imports are enough to get started, and one channel is usually enough per JS process. However you may want more than one channel, or have control over its creation:

import { Channel } from 'polyrhythm';
const channel = new Channel();
channel.trigger(...)

(In a React environment, a similar choice exists- a top-level useListener hook, or a listener bound to a channel via useChannel. React equivalents are discussed further in the polyrhythm-react repo)

To tie cancelation into your UI layer's component lifecycle (or server-side request fulfillment if in Node), call .unsubscribe() on the return value from channel.listen or channel.filter for any handlers the component set up:

// at mount
const sub = channel.on(...)..
// at unmount
sub.unsubscribe()

Lastly in a hot-module-reloading environment, channel.reset() is handy to remove all listeners, canceling their effects. Include that call early in the loading process to avoid double-registration of listeners in an HMR environment.

API

A polyrhythm app, sync or async, can be built out of 6 or fewer primitives:

• trigger - Puts an event on the event bus, and should be called at least once in your app. Generally all a UI Event Handler needs to do is call trigger with an event type and a payload.
  Example — addEventListener('click', ()=>{ trigger('timer/start') })

• filter - Adds a function to be called on every matching trigger. The filter function will be called synchronously in the call-stack of trigger, can modify its events, and can prevent events from being further handled by throwing an Error.
  For metadata — filter('timer/start', event => { event.payload.startedAt = Date.now()) })
  Validation — filter('form/submit', ({ payload }) => { isValid(payload) || throw new Error() })

• listen - Adds a function to be called on every matching trigger, once all filters have passed. Allows you to return an Observable of its side-effects, and/or future event triggerings, and configure its overlap behavior / concurrency declaratively.
  AJAX: listen('profile/fetch', ({ payload }) => get('/user/' + payload.id)).tap(user => trigger('profile/complete', user.profile))

• query - Provides an Observable of matching events from the event bus. Useful when you need to create a derived Observable for further processing, or for controlling/terminating another Observable. Example: interval(1000).takeUntil(query('user/activity'))

Observable creators

• after - Defers a function call into an Observable of that function call, after a delay. This is the simplest way to get a cancelable side-effect, and can be used in places that expect either a Promise or an Observable.
  Promise — await after(10000, () => modal('Your session has expired'))
  Observable — interval(1000).takeUntil(after(10000)) `
• concat - Combines Observables by sequentially starting them as each previous one finishes. This only works on Observables which are deferred, not Promises which are begun at their time of creation.
  Sequence — login().then(() => concat(after(9000, 'Your session is about to expire'), after(1000, 'Your session has expired')).subscribe(modal))

You can use Observables from any source in polyrhythm, not just those created with concat and after. For maximum flexibility, use the Observable constructor to wrap any async operation - and use them anywhere you need more control over the Observables behavior. Be sure to return a cleanup function from the Observable constructor, as in this session-timeout example.

=======

Assign the output to the search/results event, and specify your mode, and you're done and race-condition-free!

on(
  'search/start',
  ({ payload }) => {
    return fetch(URL + payload).then(res => res.json());
  },
  {
    mode: 'replace',
    trigger: { next: 'search/results' },
  }
);

mode:replace does what switchMap does, but with readability foremost, and without requiring you to model your app as a chained Observable, or manage Subscription objects or call .subscribe() or .unsubscribe() explicitly.

Debounced Search CodeSandbox

Example 2: Ajax Cat Fetcher (multi-mode)

Based on an XState Example showing the value of separating out effects from components, and how to be React Concurrent Mode (Suspense-Mode) safe, in XState or Polyrhythm.

Try it out - play with it! Is the correct behavior to use serial mode to allow you to queue up cat fetches, or ignore to disable new cats while one is loading, as XState does? You choose! I find having these options easily pluggble enables the correct UX to be discovered through play, and tweaked with minimal effort.

Cat Fetcher AJAX CodeSandbox

Example 3: Redux Toolkit Counter (multi-mode)

All 5 modes can be tried in the polyrhythm version of the Redux Counter Example Sandbox

Can I use Promises instead of Observables?

Recall the auto-complete example, in which you could create a new search/results event from either a Promise or Observable:

on('search/start', ({ payload }) => {
  // return Observable
  return ajax.get(URL + payload).pipe(
    tap({ results } => results)
  );
  // OR Promise
  return fetch(URL + payload).then(res => res.json())
}, {
 mode: 'replace',
 trigger: { next: 'search/results' }
});

With either the Promise, or Observable, the mode: replace guarantees your autocomplete never has the race-condition where an old result populates after new letters invalidate it. But with an Observable:

• The AJAX can be canceled, freeing up bandwidth as well
• The AJAX can be set to be canceled implicitly upon component unmount, channel reset, or by another event declaratively with takeUntil. And no Abort Controllers or await ever required!

You have to return an Observable to get cancelation, and you only get all the overlap strategies and lean performance when you can cancel. So best practice is to use them - but they are not required.

UI Layer Bindings

trigger, filter listen (aka on), and query are methods bound to an instance of a Channel. For convenience, and in many examples, these bound methods may be imported and used directly

import { trigger, on } from 'polyrhythm';
on(...)
trigger(...)

These top-level imports are enough to get started, and one channel is usually enough per JS process. However you may want more than one channel, or have control over its creation:

import { Channel } from 'polyrhythm';
const channel = new Channel();
channel.trigger(...)

(In a React environment, a similar choice exists- a top-level useListener hook, or a listener bound to a channel via useChannel. React equivalents are discussed further in the polyrhythm-react repo)

To tie cancelation into your UI layer's component lifecycle (or server-side request fulfillment if in Node), call .unsubscribe() on the return value from channel.listen or channel.filter for any handlers the component set up:

// at mount
const sub = channel.on(...)..
// at unmount
sub.unsubscribe()

Lastly in a hot-module-reloading environment, channel.reset() is handy to remove all listeners, canceling their effects. Include that call early in the loading process to avoid double-registration of listeners in an HMR environment.

API

A polyrhythm app, sync or async, can be built out of 6 or fewer primitives:

• trigger - Puts an event on the event bus, and should be called at least once in your app. Generally all a UI Event Handler needs to do is call trigger with an event type and a payload.
  Example — addEventListener('click', ()=>{ trigger('timer/start') })

• filter - Adds a function to be called on every matching trigger. The filter function will be called synchronously in the call-stack of trigger, can modify its events, and can prevent events from being further handled by throwing an Error.
  For metadata — filter('timer/start', event => { event.payload.startedAt = Date.now()) })
  Validation — filter('form/submit', ({ payload }) => { isValid(payload) || throw new Error() })

• listen - Adds a function to be called on every matching trigger, once all filters have passed. Allows you to return an Observable of its side-effects, and/or future event triggerings, and configure its overlap behavior / concurrency declaratively.
  AJAX: listen('profile/fetch', ({ payload }) => get('/user/' + payload.id)).tap(user => trigger('profile/complete', user.profile))

• query - Provides an Observable of matching events from the event bus. Useful when you need to create a derived Observable for further processing, or for controlling/terminating another Observable. Example: interval(1000).takeUntil(query('user/activity'))

Observable creators

• after - Defers a function call into an Observable of that function call, after a delay. This is the simplest way to get a cancelable side-effect, and can be used in places that expect either a Promise or an Observable.
  Promise — await after(10000, () => modal('Your session has expired'))
  Observable — interval(1000).takeUntil(after(10000)) `
• concat - Combines Observables by sequentially starting them as each previous one finishes. This only works on Observables which are deferred, not Promises which are begun at their time of creation.
  Sequence — login().then(() => concat(after(9000, 'Your session is about to expire'), after(1000, 'Your session has expired')).subscribe(modal))

You can use Observables from any source in polyrhythm, not just those created with concat and after. For maximum flexibility, use the Observable constructor to wrap any async operation - and use them anywhere you need more control over the Observables behavior. Be sure to return a cleanup function from the Observable constructor, as in this session-timeout example.

1.2.6 After can defer an Observable

listen('user/activity', () => {
  return concat(
    new Observable(notify => {       // equivalent to after(9000, "Your session is about to expire")
      const id = setTimeout(() => {
        notify.next("Your session is about to expire");
        notify.complete();           // tells `concat` we're done- Observables may call next() many times
      }, 9000);
      return () => clearTimeout(id); // a cancelation function allowing this timeout to be 'replaced' with a new one
    }),
    after(1000, () => "Your session has expired"));
}, { mode: 'replace' });
});

List Examples - What Can You Build With It?

• The Redux Counter Example
• The Redux Todos Example
• A requestAnimationFrame-based Game Loop
• Seven GUIs Solutions 1-Counter | 2-Temperature | 3-Flight | 4-CRUD | 5-Timer (more info at 7 GUIs)
• The Chat UI Example with TypingIndicator
• See All CodeSandbox Demos

FAQ

Got TypeScript typings?

But of course!

How large? 16Kb parsed size, 4Kb Gzipped

In Production Use? Yes.

What does it do sync, async? With what Error-Propogation and Cancelability? How does it work?

See The test suite for details.

How fast is it? Nearly as fast as RxJS. The Travis CI build output contains some benchmarks.

Tutorial: Ping Pong 🏓

const { filter, trigger, log } = require();

// **** Behavior (criteria, fn) ****** //
filter(true, log);

// **** Events (type, payload) ****** //
trigger('ping', 'Hello World');

// **** Effects ({ type, payload }) ****** //
function log({ type, payload }) {
  console.log(type, payload ? payload : '');
}
// Output:
// ping Hello World

const { filter, listen, log, trigger } = require('polyrhythm');

filter(true, log);
listen('ping', () => {
  trigger('pong');
});

trigger('ping');
// Output:
// ping
// pong

const { filter, listen, log, after, trigger } = require('polyrhythm');

filter(true, log);
listen('ping', () => {
  return after(1000, () => trigger('pong'));
});

trigger('ping');
// Output: (1000 msec between)
// ping
// pong

async function sneeze() {
  await after(1000, () => log('Ah..'));
  await after(1000, () => log('..ah..'));
  await after(1000, () => log('..choo!'));
}

const { filter, listen, log, after, trigger } = require('polyrhythm');

filter(true, log);
listen('ping', () => {
  return after(1000, () => trigger('pong'));
});
listen('pong', () => {
  return after(1000, () => trigger('ping'));
});

trigger('ping');
// Output: (1000 msec between each)
// ping
// pong
// ping
// pong  (etc...)

filter(true, log);
listen(['ping', 'pong'], returnBall);

trigger('ping');

function returnBall({ type }) {
  return after(1000, () => trigger(type === 'ping' ? 'pong' : 'ping'));
}

const { filter, listen, log, after, reset, trigger } = require('polyrhythm');

filter(true, log);
listen(['ping', 'pong'], returnBall);

trigger('ping');
after(4000, reset).then(() => console.log('Done!'));

//Output:
// ping
// pong
// ping
// pong
// Done!

filter(true, log);
listen('ping', () => {
  return after(1000, () => trigger('pong'));
});
const player2 = listen('pong', () => {
  return after(1000, () => trigger('ping'));
});

trigger('ping');
after(4000, () => player2.unsubscribe()).then(() => console.log('Done!'));

//Output:
// ping
// pong
// ping
// pong
// Done!

Further Reading

The following were inspiring principles for developing polyrhythm, and are definitely worth reading up on in their own right:

• Command Object Pattern
• Pub Sub Pattern
• Actor Model
• Event Sourcing / CQRS
• Flux Standard Action | Redux
• TC39 Observable proposal
• ReactiveX
• RxJS Concurrency operators
• Turning the Database Inside Out (Samza/Kafka)
• Svelte
• Elm