Revocable Queue

Revocable Queue allows you to read/write a sequence of data values (aka, a queue) asynchronously, similar to streams or observables. But any data/event that is still pending in the queue -- hasn't yet been read -- can be revoked.

Several helpers are also included to make working with revocable queues easier, using the ES2018 asynchronous iteration protocol, including: lazyZip(..), lazyMerge(..), queueIterable(..), and eventIterable(..). In addition, eventState(..) helps with managing the asynchronous toggling of event states.

API

To create a queue, call create():

var q = RevocableQueue.create();

To add values to the queue, call add(..):

q.add( Math.random() );

Revoking

The add(..) function call returns a function:

var revoke = q.add( Math.random() );

This returned function attempts to revoke the specific value you previously added to the queue; it will either revoke the value from the queue and return it, if it hasn't been read yet, or return the RevocableQueue.EMPTY value if the value has already been read:

// oops, let's try to revoke that value from the queue
var val = revoke();
if (val !== RevocableQueue.EMPTY) {
    console.log(`Random number we revoked: ${val}`);
}
else {
    console.log("Oops, too late to revoke");
}

This revoke(..) function takes one optional boolean argument, which defaults to true; the argument indicates if the value should be fully removed from the queue (if it hasn't been read yet). To essentially "peek" at the value without actually revoking it from the queue, pass false for the argument:

var peek = revoke(/*take=*/false);

console.log(`This random number (${peek}) is still in the queue.`);

Note: This revoke(..) function is the same function returned when actually reading a value from the queue (with next()). The "peek" usage just described is generally more useful in that scenario; more on that below.

The add(..) utility adds a value to the end of the queue. If you need to override the order of the queue and ensure an inserted value will be consumed next from the queue, use insertFirst(..). This utility works identically to add(..) except for where in the queue it ends up:

var revoke = q.insertFirst( Math.random() );

Reading / Taking

To read/take from a queue, call next(), which returns a promise that resolves as soon as there's a value in the queue to be read. The resolved value is a function (identical to the revoke(..) function described above) that is used to actually take a value from the queue (or just peek at it):

// using `async..await` syntax:
var get = await q.next();

var val = get();
if (val !== RevocableQueue.EMPTY) {
    console.log(`Random number taken from the queue: ${val}`);
}
else {
    console.log("Oops, value already revoked");
}

// using promise chain syntax:
var pr = q.next();

pr.then(function t(get){
    var val = get();
    if (val !== RevocableQueue.EMPTY) {
        console.log(`Random number taken from the queue: ${val}`);
    }
    else {
        console.log("Oops, value already revoked");
    }
});

Note: The read function (named get() here) is only resolved if there's a value ready at the moment. However, it' possible (but rare!) that between that moment and when get() is called, the value has already been revoked. For this reason, it's recommended to call get() as soon as it's received, to significantly reduce the chances of such a race condition. For robustness, always perform the if check as illustrated above. If the get() call returns RevocableQueue.EMPTY, the value was already revoked, and that get() function should now be discarded. Call next() on the queue again to get a promise for the next get() read function.

Peeking

Why does the promise returned from next() not just resolve to the taken value? The reason: you may want to "peek" at value before committing to taking it.

The usefulness of "peeking" is mostly if you want to synchronize taking values from two or more queues, and only want to take a value from each queue if all desired queue-values are (still) ready to take at that exact moment:

var [ get1, get2 ] = await Promise.all([ q1.next(), q2.next() ]);
// both queue values ready?
if (
    get1(/*take=*/false) !== RevocableQueue.EMPTY &&
    get2(/*take=*/false) !== RevocableQueue.EMPTY
) {
    let val1 = get1(/*take=*/true);
    let val2 = get2(/* defaults to true */);

    console.log(`The values retrieved: ${val1} ${val2}`);
}

In this above snippet, the not-so-obvious race condition is that get1 may have been resolved signficantly before (or after) get2, so by that time, either underlying value may have been revoked. The if statement peeks through each queue's ready-to-read accessor function to ensure the value is indeed still ready.

Note: There is no race condition between the get1(false) and the get1(true) call (or the two get2(..) calls), because JavaScript is single-threaded. So as long as this code pattern is followed, where the peeking and the reading happen synchronously (no promise/await deferral in between!), it's perfectly safe to assume that the peeked value is still ready to read in the next statement. Even if some other code was trying to revoke that value at that exact moment, it would be waiting for this code to finish, and since it's fully read/taken by then, the revoking would fail.

Synchronizing of lazy async reads from multiple queues is an expected common use-case for Revocable Queue; the lazyZip(..) helper utility is thus provided.

Closing a Queue

A queue can be closed at any point, which implicitly revokes and discards any unconsumed values in the queue at that point. After a queue is closed, any retained references to accessors from the queue will thereafter immediately return the RevocableQueue.EMPTY value.

A queue has an isClosed() method to check if it has been closed or not. Once a queue has been closed, avoid calling add(..) / insertFirst(..) (throws an exception) or next() (returns a rejected promise with an exception).

async function main() {
    var q = RevocableQueue.create();

    q.add( 1 );
    q.add( 2 );
    q.add( 3 );
    q.add( 4 );
    var accessor = await q.next();
    accessor(/*take=*/true);     // 1

    accessor = await q.next();
    accessor(/*take=*/false);    // 2

    q.close();
    q.isClosed();                // true
    accessor(/*take=*/true);     // RevocableQueue.EMPTY
    try {
        q.add( 5 );
    }
    catch (err) {
        err.toString();          // Error: Queue is closed
    }
    try {
        v = await q.next();
    }
    catch (err) {
        err.toString();          // Error: Queue is closed
    }
}

Example

To illustrate usage with (as expected) asynchronously produced and read values:

var q = RevocableQueue.create();
var revokes = [];

// add a new value to the queue every 100ms
setInterval(function write(){
    var revoke = q.add( Math.random() );
    revokes.push(revoke);
},100);


// elsewhere:


// read a value from the queue every 500ms
(async function readValues(){
    while (true) {
        let get = await q.next();
        if (get(/*take=*/false) !== RevocableQueue.EMPTY) {
            let val = get();
            console.log(`Read value: ${val}`);
        }

        // `delay(..)` is a promisified `setTimeout(..)`
        await delay(500);
    }
})();


// elsewhere:


clearButton.addEventListener("click",function clearQueue(){
    for (let revoke of revokes) {
        let result = revoke();
        // NOTE: `result` would be `RevocableQueue.EMPTY` if
        // the revocation failed, which here doesn't matter
    }
    revokes.length = 0;
});

The readValues(..) asynchronous looping is only reading one value per 500ms, but the setInterval(..) loop is adding a new value every 100ms. As such, the queue is going to grow by ~8 values per second. The button click handler clears out the queue by calling all the collected revoke() functions.

readValues(..) shows how to manually consume a queue, which in some cases may be necessary. But it's generally recommended to use the ES2018 asynchronous iteration pattern, which is illustrated later.

Iterating Queues

Asynchronous iteration (ES2018) is a language-standard pattern for consuming an async stream of values. Revocable Queue provides helpers for consuming queues as async iterators, commonly with a for await.. loop.

`lazyZip(..)`

In the Peeking section above, the use case was presented to synchronize lazy asynchronous reads from multiple queues. For example, imagine a queue that collects customers who attempt to connect to a support chat, and another queue that collects support agents who become available for a chat.

You would want to read a value from each queue, whenever one was ready, but ensure that you have an atomic pairing, meaning that you only take a value from each queue when both queues have a value available. If a customer starts waiting, and later an agent becomes available, you pair them. But if the customer disconnects while waiting for an agent, they need to be removed (revoked) from the queue. The reverse scenario also applies: an agent could disconnect while waiting for a customer to connect.

In FP (functional programming) terms, as well as in reactive/streams based programming like with observables, this atomic pairing is most closely modeled as a zip(..) operation. But traditional FP zip(..) utilities don't account for reading from queues which are asynchronous, and observables' zip(..) doesn't really account for revocation. Neither is exactly the model we need.

Since Revocable Queue queues hold values that can be revoked, this lazyZip(..) helper utility has the nuanced but important difference from other "zip"s that it doesn't actually consume a value from a queue until all queues have a value available to consume. Any queue that's being iterated with lazyZip(..) can have a previous value revoked if it hasn't yet actually been consumed by lazyZip(..). Also, that queue may be closed, which implicity revokes/discards any unconsumed values.

Here's how to use lazyZip(..):

var agents = RevocableQueue.create();
var customers = RevocableQueue.create();

async function supporChatSessions() {
    for await (
        let [customer,agent] of RevocableQueue.lazyZip(customers,agents)
    ) {
        console.log(`Chat session ready for ${customer} and ${agent}`);
    }
}

lazyZip(..) returns an ES2018 async iterator instance (suitable for use with for await.. loops, etc). This makes consumption of the resulting stream (aka "queue") of value collections straightforward.

As is common for a "zip" operation, the iterator from lazyZip(..) will be closed if any of the queues are closed (which would terminate the for await.. loop). Also, the iterator will be closed if you early-exit a for await.. loop (uncaught exception, break, return, etc), or if you directly call return() on an iterator instance.

Note: Closing an iterator does not close any of its observed queues, but closing any such queue will close the iterator.

`lazyMerge(..)`

You may want to consume the next available value from any of multiple queues, which in reactive programming speak is a "merge". lazyMerge(..) returns an ES2018 async iterator instance to make this consumption straightforward:

var clicks = RevocableQueue.create();
var keypresses = RevocableQueue.create();

async function moveGameCharacter() {
    for await (
        let evt of RevocableQueue.lazyMerge(clicks,keypresses)
    ) {
        if (evt.type == "click") {
            // ..
        }
        else if (evt.type == "keypress") {
            // ..
        }
    }
}

As is common for a "merge" operation, the iterator from lazyMerge(..) will be closed if all queues are closed (which would terminate the for await.. loop). Also, the iterator will be closed if you early-exit a for await.. loop (uncaught exception, break, return, etc), or if you directly call return() on an iterator instance.

Note: Closing an iterator does not close any of its observed queues, but closing all such queues will close the iterator.

Iterating a Single Queue

lazyZip(..) and lazyMerge(..) are primarily intended to zip/merge values from two or more revocable queues. However, you can pass either utility a single revocable queue, and what you get back is an async iterator to consume that single queue asynchronously in a language-standard, convenient form.

To illustrate, recall this manual iteration example from above:

// read a value from the queue every 500ms
(async function readValues(){
    while (true) {
        let get = await q.next();
        if (get(/*take=*/false) !== RevocableQueue.EMPTY) {
            let val = get();
            console.log(`Read value: ${val}`);
        }

        // `delay(..)` is a promisified `setTimeout(..)`
        await delay(500);
    }
})();

Here's how to consume that queue using lazyMerge(..):

// read a value from the queue every 500ms
(async function readValues(){
    for await (let val of RevocableQueue.lazyMerge(q)) {
        console.log(`Read value: ${val}`);

        // `delay(..)` is a promisified `setTimeout(..)`
        await delay(500);
    }
})();

Note: lazyZip(..) can also be used for this purpose, but it always yields an array, even with a single value. So, the for await.. statement above would be: for await (let [val] of RevocableQueue.lazyZip(q)), with the [val] array destructuring.

The semantic of using lazyMerge(..) / lazyZip(..) here with a single queue may be a bit confusing to readers of your code. So it's recommended to use the queueIterable(..) helper, which is solely provided as a better semantic name for this use-case (it just invokes lazyMerge(..) with a single passed in queue):

// read a value from the queue every 500ms
(async function readValues(){
    for await (let val of RevocableQueue.queueIterable(q)) {
        console.log(`Read value: ${val}`);

        // `delay(..)` is a promisified `setTimeout(..)`
        await delay(500);
    }
})();

This form of queue iteration with queueIterable(..) is much nicer than doing it manually!

`eventIterable(..)`

A common pattern is to turn a stream of events into an async iterable for consumption. You could do this manually by creating a queue, subscribing the event to push event objects into the queue, then turning the queue into an iterable with queueIterable(..). But that's a fair amount of boilerplate.

For this purpose, the eventIterable(..) helper is provided:

var btn = ..;

(async function logButtonClicks(){
    for await (let evt of RevocableQueue.eventIterable(btn,"click")) {
        console.log(`Clicked on: ${evt.target}`);
    }
})();

This loop will continue perpetually because nothing will close the iterator or the underlying queue from the event subscription. However, if the iterator is closed, the event is unsubscribed during cleanup.

`eventState(..)`

Another use-case for revocable queues and lazyZip(..) is listening for alternating events to fire that represent a toggling of a state (between true and false). The concern is not receiving specific values from these events (as illustrated previously with lazyZip(..)), but rather just listening for a signal that all of the activation events for a set of two or more listeners has fired, and that no corresponding deactivation events occurred while waiting.

For example: managing a series of network socket connections which fire "connected" and "disconnected" events, and synchronizing operations to occur only when all the connections are active/connected at the same time.

For this kind of event/state synchronization use case, eventState(..) is provided, which wraps lazyZip(..) and subscribes to events on EventEmitter-compatible event emitter instances (ie, .on(..) for subscribing and .off(..) for unsubscribing).

Note: eventState(..) will typically be used with full event emitter instances (from streams, network sockets, etc). But if you need to create event emitters directly, see EventEmitter() below.

To use eventState(..), pass it an array of two or more objects. Each object should have at a minimum a listener property with the event emitter instance, as well as an onEvent property with the name of the activation event to listen for (or an array of event names).

Optionally, each of these objects can include an offEvent property to name a deactivation event (or array of event names) to listen for, and a status property (boolean, default: false) to initialize the status for each listener.

The return value from eventState(..) is a controller object, which has two properties: wait is the promise that will resolve when the event-state synchronization has completed, and cancel() which will cancel the processing (including cleaning up all event handlers) and cause the wait promise to be rejected.

async function greetings(conn1,conn2,conn3) {
    var controller = RevocableQueue.eventState([
        {
            listener: conn1,
            onEvent: [ "connected", "reconnected", ],
            offEvent: "disconnected",
            status: conn1.isConnected
        },
        {
            listener: conn2,
            onEvent: [ "connected", "reconnected", ],
            offEvent: "disconnected",
            status: conn2.isConnected
        },
        {
            listener: conn3,
            onEvent: [ "connected", "reconnected", ],
            offEvent: "disconnected",
            status: conn3.isConnected
        }
    ]);

    // listen for an "abandon" event to cancel the event-state processing
    conn1.on("abandon",controller.cancel);

    try {
        await controller.wait;
        broadcastMessage( [conn1,conn2,conn3], "greetings!" );
    }
    catch (err) {
        console.log("Connections abandoned!");
    }
}

This code asserts that the three network socket connection objects (conn1, conn2, and conn3) all emit "connected" / "reconnected" and "disconnected" events, as well as have an isConnected boolean property that's true when connected or false when not. The moment all 3 connections are established simultaneously, the await expression waiting on the wait promise will complete, and then the broadcastMessage(..) operation will be performed.

If the "abandon" event is fired on conn1, controller.cancel() is invoked, which cancels the event-state processing, cleans up all the listeners, and then rejects the wait promise.

`EventEmitter`

RevocableQueue.EventEmitter() is an included utility to create simple event emitter instances:

var listener = new RevocableQueue.EventEmitter();

listener.on("greeting",function onHello(msg){
    console.log(`Hello, ${msg}!`);
});

listener.emit("greeting","Kyle");
// Hello, Kyle!

RevocableQueue.EventEmitter() is a stripped-down implementation of a synchronous event emitter, with a generally compatible subset of the Node.js EventEmitter() API.

Note: Only use if your code's environment doesn't already provide a suitable event emitter utility.

RevocableQueue.EventEmitter() instances have the following methods:

emit(eventName,...data)
on(eventName,handler)
once(eventName,handler)
removeListener(eventName,handler)
off(eventName,handler)
removeAllListeners(eventName) (note: eventName is optional)

Builds

The distribution library file (dist/rq.js) comes pre-built with the npm package distribution, so you shouldn't need to rebuild it under normal circumstances.

However, if you download this repository via Git:

The included build utility (build-core.js) builds (and minifies) dist/rq.js from source. The build utility expects Node.js version 6+.
To install the build and test dependencies, run npm install from the project root directory.
- Note: This npm install has the effect of running the build for you, so no further action should be needed on your part.
To manually run the build utility with npm:
```
npm run build
```
To run the build utility directly without npm:
```
node build-core.js
```

Tests

A test suite is included in this repository, as well as the npm package distribution. The default test behavior runs the test suite using index.js.

The included Node.js test utility (node-tests.js) runs the test suite. This test utility expects Node.js version 6+.
To run the test utility with npm:
```
npm test
```
Other npm test scripts:
- npm run test:dist will run the test suite against dist/rq.js instead of the default of index.js.
- npm run test:package will run the test suite as if the package had just been installed via npm. This ensures package.json:main properly references dist/rq.js for inclusion.
- npm run test:all will run all three modes of the test suite.
To run the test utility directly without npm:
```
node node-tests.js
```

Test Coverage

If you have NYC (Istanbul) already installed on your system (requires v14.1+), you can use it to check the test coverage:

npm run coverage

Then open up coverage/lcov-report/index.html in a browser to view the report.

Note: The npm script coverage:report is only intended for use by project maintainers. It sends coverage reports to Coveralls.

getify/revocable-queue