npm package discovery and stats viewer.

Discover Tips

  • General search

    [free text search, go nuts!]

  • Package details

    pkg:[package-name]

  • User packages

    @[username]

Sponsor

Optimize Toolset

I’ve always been into building performant and accessible sites, but lately I’ve been taking it extremely seriously. So much so that I’ve been building a tool to help me optimize and monitor the sites that I build to make sure that I’m making an attempt to offer the best experience to those who visit them. If you’re into performant, accessible and SEO friendly sites, you might like it too! You can check it out at Optimize Toolset.

About

Hi, 👋, I’m Ryan Hefner  and I built this site for me, and you! The goal of this site was to provide an easy way for me to check the stats on my npm packages, both for prioritizing issues and updates, and to give me a little kick in the pants to keep up on stuff.

As I was building it, I realized that I was actually using the tool to build the tool, and figured I might as well put this out there and hopefully others will find it to be a fast and useful way to search and browse npm packages as I have.

If you’re interested in other things I’m working on, follow me on Twitter or check out the open source projects I’ve been publishing on GitHub.

I am also working on a Twitter bot for this site to tweet the most popular, newest, random packages from npm. Please follow that account now and it will start sending out packages soon–ish.

Open Software & Tools

This site wouldn’t be possible without the immense generosity and tireless efforts from the people who make contributions to the world and share their work via open source initiatives. Thank you 🙏

© 2024 – Pkg Stats / Ryan Hefner

queueable

v5.3.2

Published

Convert push-based streams to pull-based async iterables

Downloads

6,080

Readme

A library for converting push-based asynchronous streams like node streams or EventTarget to pull-based streams implementing the ES2018 asynchronous iteration protocols.

Well-typed, well-tested and lightweight.

Overview

Asynchronous iteration is a new native feature of JavaScript for modeling streams of values in time. To give a rough analogy, asynchronous iteration is to event emitters as promises are to callbacks. The problem this library helps to solve is that iterables have a pull-based interface, while sources like event emitters are push-based, and converting between the two kinds of providers involves buffering the difference between pushes and pulls until it can be settled. This library provides push-pull adapters with buffering strategies suitable for different use cases.

Queueable is intended both for library authors and consumers. Library authors can implement a standard streaming interface for interoperability, and consumers can adapt not-yet interoperable sources to leverage tools like IxJS and a declarative approach to streams.

Similarity to Streams API and node streams

Asynchronous iteration together with this library could be seen as a lightweight version of the WHATWG Streams API. Specifically, the adapters work like identity transform streams. Asynchronous iteration has been added to Streams API ReadableStream.

Node streams have already implemented asynchronous iteration for reading.

The use-case for this library, given that there are more standard alternatives, is based on its small size. Older browsers and node versions don't implement the newer APIs, and including a polyfill for a large API can be prohibitive.

Similarity to CSP channels

Communicating sequential processes (CSP) is a concurrency model used in Go goroutines and Clojure's core.async that is based on message passing via channels, and it's been possible to express this model in JavaScript with ES6 generators, as shown by js-csp. Asynchronous iteration brings JavaScript closer to having first-class syntactical support of channels, as can be seen in this demonstration of ping-pong adapted from Go and js-csp using Queuable.

Use cases

Sources of asynchronous data that are pull-based (are backpressurable; allow the consumer to control the rate at which it receives data) are trivial to adapt to asynchronous iterators using asynchronous generator functions. Such sources include event emitters that can be paused and resumed, and callback functions that are fired a single time, and functions that return promises.

Converting pull-based sources to asynchronous iterables is still made easier by the wrapRequest helper method provided by this library. For a demonstration, see requestAnimationFrame example (also showing IxJS usage) and implementing an example interval.

Sources that are not backpressurable can only be sampled by subscribing to them or unsubscribing, and examples of such sources are user events like mouse clicks. Users can't be paused, so this library takes care of buffering the events they generate until requested by the consumer. See mouse events demonstration.

Asynchronous iteration

See slides about Why Asynchronous Iterators Matter for a more general introduction to the topic.

Installation

npm install --save queueable
yarn add queueable

CDN

https://unpkg.com/queueable/dist/queueable.umd.js

Adapters

Channel

Push-pull adapter backed by unbounded linked list queues (to avoid array reindexing) with optional circular buffering.

Circular buffering works like a safety valve by discarding the oldest item in the queue when the limit is reached.

Methods

  • static constructor(pushLimit = 0, pullLimit = 0)
  • static fromDom(eventType, target[, options])
  • static fromEmitter(eventType, emitter)
  • push(value, [done]) Push a value to the queue; returns a promise that resolves when the value is pulled.
  • wrap([onReturn]) Return an iterable iterator with only the standard methods.

Examples

Implementing an asynchronous iterable iterator, pushing values to it and then consuming with for-await-of
import { Channel } from 'queueable';

const channel = new Channel();
channel.push(1);
channel.push(2);
channel.push(3);
channel.push(4, true); // the second argument closes the iterator when its turn is reached

// for-await-of uses the async iterable protocol to consume the queue sequentially
for await (const n of channel) {
  console.log(n); // logs 1, 2, 3
  // doesn't log 4, because for-await-of ignores the value of a closing result
}
// the loop ends after it reaches a result where the iterator is closed
Pulling results and waiting for values to be pushed
const channel = new Channel();
const result = channel.next(); // a promise of an iterator result
result.then(({ value }) => {
  console.log(value);
});
channel.push('hello'); // "hello" is logged in the next microtick
Hiding the adapter methods from consumers with wrap()

The iterables should be one-way for end-users, meaning that the consumer should only be able to request values, not push them, because the iterables could be shared. The wrap([onReturn]) method returns an object with only the standard iterable methods.

This example adapts an EventTarget in the same way as the fromDom() method.

const channel = new Channel();
const listener = (event) => void channel.push(event);
eventTarget.addEventListener('click', listener);
const clickIterable = channel.wrap(() => eventTarget.removeEventListener(type, listener));
clickIterable.next(); // -> a promise of the next click event
clickIterable.return(); // closes the iterable
Tracking when pushed values are pulled

The push() methods for the adapters return the same promise as the next() methods for the iterators, so it's possible for the provider to track when the pushed value is used to resolve a pull.

const channel = new Channel();
const tracking = channel.push(123);
tracking.then(() => {
  console.log('value was pulled');
});
const result = channel.next(); // pulling the next result resolves `tracking` promise
result === tracking; // -> true
(await result) === (await tracking); // -> true

LastResult

An adapter that only buffers the last value pushed and caches and broadcasts it (pulling a value doesn't dequeue it). It's suitable for use cases where skipping results is allowed.

Methods

  • static constructor()
  • static fromDom(eventType, target[, options])
  • static fromEmitter(eventType, emitter)
  • push(value) Overwrite the previously pushed value.
  • wrap([onReturn]) Return an iterable iterator with only the standard methods.

Examples

Converting mouse move events into a stream
import { LastResult } from 'queueable';
const moveIterable = LastResult.fromDom('click', eventTarget);
for await (const moveEvent of moveIterable) {
  console.log(moveEvent); // logs MouseEvent objects each time the mouse is clicked
}
// the event listener can be removed and stream closed with .return()
moveIterable.return();

wrapRequest(request[, onReturn])

The wrapRequest() method converts singular callbacks to an asynchronous iterable and provides an optional hook for cleanup when the return() is called.

Examples

Adapting requestAnimationFrame()
const { wrapRequest } = 'queueable';
const frames = wrapRequest(window.requestAnimationFrame, window.cancelAnimationFrame);
for await (const timestamp of frames) {
  console.log(timestamp); // logs frame timestamps sequentially
}
Creating an iterable interval with setTimeout()
const makeInterval = (delay) =>
  wrapRequest((callback) => window.setTimeout(callback, delay), window.clearTimeout);
const interval = makeInterval(100); // creates the interval but does nothing until .next() is invoked
let i = 0;
for await (const _ of interval) {
  i += 1;
  if (i === 10) {
    interval.return(); // stops the interval
  }
}

Multicast

The same concept as Subject in observables; allows having zero or more subscribers that each receive the pushed values. The pushed values are discarded if there are no subscribers. Uses the Channel adapters internally.

import { Multicast } from 'queueable';

const queue = new Multicast();
// subscribe two iterators to receive results
const subscriberA = queue[Symbol.asyncIterator]();
const subscriberB = queue[Symbol.asyncIterator]();
queue.push(123);
const results = Promise.all([subscriberA.next(), subscriberB.next()]);
console.log(await results); // logs [{ value: 123, done: false }, { value: 123, done: false }]

Types

To make TypeScript know about the asnyc iterable types (AsyncIterable<T>, AsyncIterator<T>, AsyncIterableiterator<T>), the TypeScript --lib compiler option should include "esnext.asynciterable" or "esnext".

Alternatives

Tools for async iteration

  • IxJS – supports various combinators for async iterables
  • Symbola – protocol extension based combinators for async iterables
  • Axax – async iteration helpers
  • iterall – iteration utilities
  • iter-tools – iteration helpers