Qottle recipe: How to manage an asynchronous polling queue

 

Qottle is a queue for asynchronous tasks with prioritization, ratelimit, and concurrency support. This recipe shows an example of a qottle queue for continuous, controlled polling.

Page Content hide
1 Qottle recipe for polling
1.1 Alternative and simpler approach

Qottle recipe for polling

You can use qottle to manage an endless, or constrained polling queue. In this scenario, we want to poll an aynch API a maximum of 100 times, but no more than 5 times every 10 seconds, and only 1 call at a time.

set up the queue

  const q = new Qottle({
    // polling every 1 seconds
    concurrent: 1,
    rateLimited: true,
    rateLimitPeriod: 10 * 1000,
    rateLimitMax: 5
  });

`
the number of iterations (or Infinite for ever)

  const ITERATIONS = 100

`
This is where you’d make the async api call – For simulation as here, qottle has a handy timer you can use for timeouts as promises which just waits for a while then returns how long it waited. You could handle the results, or errors here, or use the qottle finish and error events.

  const action = () => q.timer(Math.floor(Math.random() * 2000));

`
handle the results of each poll – you could do this on item resolution of q.add, or by using the finish event – as here, where polling results are just being added to an array

  const results = [];
  q.on("finish", ({ entry, result }) => {
    results.push({
      entry,
      result,
    });
  });

`
create a recursive function for adding stuff to the queue – this one should work for most situations. It will finally resolve when the number of iterations are reached.

  const adder = ({ entry, result } = { entry: { key: 0 } }) =>
    q
      .add(() => action(), { key: entry.key + 1 })
      .then(({ entry, result }) =>
        entry.key < ITERATIONS
          ? adder({ entry, result })
          : Promise.resolve({ entry, result })
      );


For testing, at the end, I’m checking that the final result and number of items processed is as expected.

  return adder().then(({ entry, result }) => {
    t.is(results.length, entry.key);
    t.is(results.length, ITERATIONS);
  });

Alternative and simpler approach

That example, had a recursive approach, where a promise resolution caused a new queue addition. Personaly I prefer this approach, but is a litle difficult to get your head around. Another (simpler) method could be to use the finish event.
 
The adder can be simpler too

  const adder = ({ entry, result } = { entry: { key: 0 } }) =>
    q.add(() => action(), { key: entry.key + 1 })
And just use the finish event to call it again
 
  q.on('finish', (({entry, result})=> {
    // optionally check here for whether it should finish
    adder ({entry, result})
  })

A full version of this example is in test.js at the Qottle repo

See the Qottle documentation for more information. Qottle  source code is on github and can be installed with npm or yarn