In this followup post I take this approach to a whole new level, showing how it can be extended to not only handle collections of items but also sequences of events. In other words, it can iterate both over containers of elements in (memory) space, and over elements generated dynamically in time.
Introducing Asynchronous Iterators and Generators
Alternatively, I can implement this using the library described in the previous post, as:
Internally, that library uses the exact same iteration mechanism.
But suppose that instead of a collection of items, I need to process a sequence of events that are generated sequentially, one after the other. And that I don’t want to wait until all the events are generated; rather, I want to process them immediately as they arrive. For example, such events might be user interactions with a UI element, such as clicks on a button, or data received over the network, a packet at a time.
I want to be able to handle such data in the same way as items in a collection. In the past this required using third-party libraries, such as RxJS. And while such libraries provide a lot of power, they can also have significant overhead, both in terms of code size and complexity, and in terms of cognitive overhead required to learn them.
Asynchronous iterators work like regular iterators, with a few notable differences:
The asynchronous iterator instance is obtained by invoking the [Symbol.asyncIterator] method, instead of [Symbol.iterator]
Rather than returning an object that has a value and done properties, as per the iteration protocol, the next method of an asynchronous iterator returns a promise, which resolves to an object that has these properties
As an implicit aspect of the asynchronous iteration protocol, the next promise is not requested until the previous one resolves
Assuming object sequence implements asynchronous iteration, we can process the events that flow through it, as they arrive, using the following simple code:
This code is almost identical to the code at the beginning of this section, with the minor addition of the await keyword. But the behavior is very different in that the code inside the loop will be invoked asynchronously, as events arrive.
Here is an example of how this mechanism can be used to count from 0 to 9, with a delay of one second between each output:
This code is low-level in that the asynchronous iterator is created and managed explicitly. An easier approach would be to use an async generator to create such an iterator, as it’s easier to use a regular generator for creating synchronous iterators, for example:
An extra benefit of using for await is that if the specified object we are iterating over doesn’t implement the [Symbol.asyncIterator] method, for await will automatically try to fallback to [Symbol.iterator], and use a synchronous iterator instead.
In a scenario like that, for await behaves like a regular for of. This means that the same iteration code can be used for both asynchronous sequences and regular collections.
The Case For Asynchronous Iteration Methods
Having said all that, what we really want is to be able to use iteration methods for, is to process the values provided by an asynchronous iterator - for the same reasons that we prefer to use iteration methods for processing collections.
As with regular iteration methods, the benefits are that you can deconstruct complex operations into a sequence of simple operations, and that each such operation is specified in a declarative manner: what you want to achieve, rather than the details of how to achieve it.
Implementing The First Asynchronous Iteration Methods
Some of the most basic iteration methods are map, filter, and forEach. Here is the implementation of these functions in the synchronous library described in the previous post:
Essentially, the only changes required are the async keyword in front of the function declarations, and the await keyword after for.
And because these methods all use for await, they also work when src is a regular container, like Array, Map, or String. This is because, as explained above, for await falls back to using regular iterators if asynchronous iterators are unavailable.
This means that this library can actually be used as a replacement for the library presented in my previous post: a single library that can handle both synchronous and asynchronous scenarios.
Creating Asynchronous Event Emitters
It’s great that we’ve so easily created methods that can operate on event sequences, when implemented using asynchronous iterators. But where do the events originally come from? And how do we create asynchronous iterators for them? At the beginning of this post I provided two examples of such event emitters - timedSequence and timedGen - but both were highly specialized for a very simple scenario.
What we need is a mechanism for transforming almost any event source into an emitter which can provide asynchronous iterators.
Here is the most complicated bit of code in this blog post, which converts a function that uses a callback to provide a sequence of values, into a sequence that supports asynchronous iteration:
The emitter argument is the function that receives a callback and generates the values that will then be provided via the asynchronous iterator, by invoking the callback with values as arguments, one at a time.
To handle a scenario in which values are generated by emitter faster than they can be consumed via the iterator, they are pushed into the values array, which functions as a temporary storage. The promise valuesAvailable is used to indicate when there are values available in the temporary storage.
These values are then yielded one-at-a-time through the asynchronous iterator to a consumer, using the yield* instruction. (yield* is a useful helper that iterates over its operand - in this case the array of pending values - and yields each value returned by it.)
One tricky bit of this code is the initialization of the valuesAvailable promise. The init function passed to the promise constructor saves the resolve callback of the promise (provided as the argument r) into the resolve variable. It also clears the temporary storage values, by resetting it to an empty array.
With fromCallback function in hand, we can easily create an asynchronous generator from a DOM event listener:
And now we can manipulate a stream of events in a functional and readable manner, for example:
This code will output the number of times the button has been clicked so far. As an exercise, see if you can create an implementation of the reduce method, in addition to filter and map implemented above. (You can find an implementation of reduce in the CodePen that I link to below.)
You can find the complete library implementation in this CodePen. Coupled with some demo code, it’s still less than 160 lines of unminified code! The complete library includes some additional functionality that I will cover in an upcoming post, such as the ability to process events from the same source through multiple streams (like multiple subscribers on a single observable), and how to connect this library to Readable Streams, such as those returned by the Fetch API through the body property of a response object.