\n\nA library for converting push-based asynchronous streams like node streams or EventTarget to pull-based streams implementing the [ES2018 asynchronous iteration protocols][async].\n\nWell-typed, well-tested and [lightweight].\n\n## Overview\n\nAsynchronous 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.\n\nQueueable 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.\n\n### Similarity to Streams API and node streams\n\nAsynchronous iteration together with this library could be seen as a lightweight version of the WHATWG [Streams API][streams]. Specifically, the adapters work like [identity transform streams][id]. Asynchronous iteration [has been added][streams-reader] to Streams API `ReadableStream`.\n\nNode streams have [already implemented][node-streams] asynchronous iteration for reading.\n\nThe 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.\n\n### Similarity to CSP channels\n\n[Communicating sequential processes][csp] (CSP) is a concurrency model used in Go goroutines and [Clojure's core.async][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][ping-pong] adapted from Go and js-csp using Queuable.\n\n[csp]: https://en.wikipedia.org/wiki/Communicating_sequential_processes\n[core.async]: https://clojure.org/news/2013/06/28/clojure-clore-async-channels\n\n### Use cases\n\nSources 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.\n\nConverting 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.\n\nSources 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.\n\n\n\n### Asynchronous iteration\n\nSee slides about [Why Asynchronous Iterators Matter][slides] for a more general introduction to the topic.\n\n## Installation\n\n```sh\nnpm install --save queueable\n```\n\n```sh\nyarn add queueable\n```\n\n### CDN\n\nhttps://unpkg.com/queueable/dist/queueable.umd.js\n\n## Adapters\n\n### [`Channel`][channel]\n\nPush-pull adapter backed by unbounded linked list queues (to avoid array reindexing) with optional circular buffering.\n\nCircular buffering works like a safety valve by discarding the oldest item in the queue when the limit is reached.\n\n#### Methods\n\n- `static constructor(pushLimit = 0, pullLimit = 0)`\n- `static fromDom(eventType, target[, options])`\n- `static fromEmitter(eventType, emitter)`\n- `push(value, [done])`\n Push a value to the queue; returns a promise that resolves when the value is pulled.\n- `wrap([onReturn])`\n Return an iterable iterator with only the standard methods.\n\n#### Examples\n\n##### Implementing an asynchronous iterable iterator, pushing values to it and then consuming with `for-await-of`\n\n```js\nimport { Channel } from 'queueable';\n\nconst channel = new Channel();\nchannel.push(1);\nchannel.push(2);\nchannel.push(3);\nchannel.push(4, true); // the second argument closes the iterator when its turn is reached\n\n// for-await-of uses the async iterable protocol to consume the queue sequentially\nfor await (const n of channel) {\n console.log(n); // logs 1, 2, 3\n // doesn't log 4, because for-await-of ignores the value of a closing result\n}\n// the loop ends after it reaches a result where the iterator is closed\n```\n\n##### Pulling results and waiting for values to be pushed\n\n```js\nconst channel = new Channel();\nconst result = channel.next(); // a promise of an iterator result\nresult.then(({ value }) => {\n console.log(value);\n});\nchannel.push('hello'); // \"hello\" is logged in the next microtick\n```\n\n##### Hiding the adapter methods from consumers with `wrap()`\n\nThe 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.\n\nThis example adapts an EventTarget in the same way as the `fromDom()` method.\n\n```js\nconst channel = new Channel();\nconst listener = (event) => void channel.push(event);\neventTarget.addEventListener('click', listener);\nconst clickIterable = channel.wrap(() => eventTarget.removeEventListener(type, listener));\nclickIterable.next(); // -> a promise of the next click event\nclickIterable.return(); // closes the iterable\n```\n\n##### Tracking when pushed values are pulled\n\nThe `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.\n\n```js\nconst channel = new Channel();\nconst tracking = channel.push(123);\ntracking.then(() => {\n console.log('value was pulled');\n});\nconst result = channel.next(); // pulling the next result resolves `tracking` promise\nresult === tracking; // -> true\n(await result) === (await tracking); // -> true\n```\n\n### [`LastResult`][lastresult]\n\nAn 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.\n\n#### Methods\n\n- `static constructor()`\n- `static fromDom(eventType, target[, options])`\n- `static fromEmitter(eventType, emitter)`\n- `push(value)`\n Overwrite the previously pushed value.\n- `wrap([onReturn])`\n Return an iterable iterator with only the standard methods.\n\n#### Examples\n\n##### Converting mouse move events into a stream\n\n```js\nimport { LastResult } from 'queueable';\nconst moveIterable = LastResult.fromDom('click', eventTarget);\nfor await (const moveEvent of moveIterable) {\n console.log(moveEvent); // logs MouseEvent objects each time the mouse is clicked\n}\n// the event listener can be removed and stream closed with .return()\nmoveIterable.return();\n```\n\n### `wrapRequest(request[, onReturn])`\n\nThe `wrapRequest()` method converts singular callbacks to an asynchronous iterable and provides an optional hook for cleanup when the `return()` is called.\n\n#### Examples\n\n##### Adapting `requestAnimationFrame()`\n\n```js\nconst { wrapRequest } = 'queueable';\nconst frames = wrapRequest(window.requestAnimationFrame, window.cancelAnimationFrame);\nfor await (const timestamp of frames) {\n console.log(timestamp); // logs frame timestamps sequentially\n}\n```\n\n##### Creating an iterable interval with `setTimeout()`\n\n```js\nconst makeInterval = (delay) =>\n wrapRequest((callback) => window.setTimeout(callback, delay), window.clearTimeout);\nconst interval = makeInterval(100); // creates the interval but does nothing until .next() is invoked\nlet i = 0;\nfor await (const _ of interval) {\n i += 1;\n if (i === 10) {\n interval.return(); // stops the interval\n }\n}\n```\n\n### [`Multicast`][multicast]\n\nThe 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.\n\n```js\nimport { Multicast } from 'queueable';\n\nconst queue = new Multicast();\n// subscribe two iterators to receive results\nconst subscriberA = queue[Symbol.asyncIterator]();\nconst subscriberB = queue[Symbol.asyncIterator]();\nqueue.push(123);\nconst results = Promise.all([subscriberA.next(), subscriberB.next()]);\nconsole.log(await results); // logs [{ value: 123, done: false }, { value: 123, done: false }]\n```\n\n## Types\n\nTo make TypeScript know about the asnyc iterable types (`AsyncIterable`, `AsyncIterator`, `AsyncIterableiterator`), the TypeScript `--lib` [compiler option][options] should include `\"esnext.asynciterable\"` or `\"esnext\"`.\n\n## Alternatives\n\n- [callback-to-async-iterator]\n- [event-iterator]\n- [Repeater.js]\n- [rxjs-for-await](https://github.com/benlesh/rxjs-for-await)\n\n## Tools for async iteration\n\n- [IxJS] – supports various combinators for async iterables\n- [Symbola] – protocol extension based combinators for async iterables\n- [Axax] – async iteration helpers\n- [iterall] – iteration utilities\n- [iter-tools] – iteration helpers\n\n[symbola]: https://github.com/slikts/symbola\n[ixjs]: https://github.com/ReactiveX/IxJS#asynciterable\n[callback-to-async-iterator]: https://github.com/withspectrum/callback-to-async-iterator\n[async]: http://2ality.com/2016/10/asynchronous-iteration.html\n[options]: https://www.typescriptlang.org/docs/handbook/compiler-options.html\n[lightweight]: https://bundlephobia.com/result?p=queueable\n[axax]: https://github.com/jamiemccrindle/axax\n[iterall]: https://github.com/leebyron/iterall\n[iter-tools]: https://github.com/sithmel/iter-tools\n[streams]: https://developer.mozilla.org/en-US/docs/Web/API/Streams_API\n[channel]: https://slikts.github.io/queueable/classes/Channel.html\n[multicast]: https://slikts.github.io/queueable/classes/multicast.html\n[lastresult]: https://slikts.github.io/queueable/classes/LastResult.html\n[slides]: https://docs.google.com/presentation/d/1r2V1sLG8JSSk8txiLh4wfTkom-BoOsk52FgPBy8o3RM\n[id]: https://streams.spec.whatwg.org/#ts\n[streams-reader]: https://github.com/whatwg/streams/issues/778\n[node-streams]: https://nodejs.org/api/stream.html#stream_readable_symbol_asynciterator\n[js-csp]: https://github.com/ubolonton/js-csp\n[ping-pong]: https://codepen.io/slikts/pen/yRPgQE?editors=0012\n[animation]: https://codepen.io/slikts/pen/mzqKvo?editors=0010\n[async-csp]: https://github.com/dvlsg/async-csp\n[event-iterator]: https://github.com/rolftimmermans/event-iterator\n[repeater.js]: https://repeater.js.org/\n","readmeFilename":"README.md"}
![](\"https://i.imgur.com/gjLSYAr.png\")
![\"Queueable\"](\"https://i.imgur.com/Ks7oik8.png\")
![\"Queueable\"](\"https://raw.githubusercontent.com/slikts/queueable/master/logo.svg?sanitize=true\"){kind=logo}
![\"License\"](\"https://img.shields.io/github/license/slikts/queueable.svg\")
\n![\"Build](\"https://img.shields.io/travis/slikts/queueable/master.svg\")
\n![\"Coverage](\"https://coveralls.io/repos/github/slikts/queueable/badge.svg?branch=master\"){kind=icon}
\n![\"Latest](\"https://img.shields.io/npm/v/queueable.svg\")
\n![\"Code](\"https://codeclimate.com/github/slikts/queueable.svg\")