async-generator.js

// Space out a series of function calls to some minimum duration.
//
// We accomplish this by first providing a way to convert a series of function calls
// into an async iterable using `asyncGeneratorOf`. 
// The resulting object also has an `add` method which is what the user calls
// to add items to the iterable.

// It then becomes a simple matter to write `spaceOut`, which simply inserts
// a wait inside the `for await` loop.

function deferred() {
  let resolve, reject;
  const promise = new Promise((res, rej) => [resolve, reject] = [res, rej];

  return {promise, resolve, reject};
}

function wait(ms) {
  return new Promise(resolve => setTimeout(resolve, ms));
}

// Create an object which can be consumed as an async iterable,
// with the ability to add items via a function call.
function asyncGeneratorOf() {
  // Maintain list of items sent in via function call,
  // not yet consumed.
  const queued = [];
  
  // Has the consumer requested an item which is not available yet?
  // In that case, return a "deferred" object stored in this variable
  // to be resolved when the next function call comes in.
  let waiting;
  
  // Remember if the user has indicated they are done sending in items.
  // This is different from the iterable being done, because
  // there could still be items waiting to be consumed.
  let done = false;

  return {
    [Symbol.iterator]() {
      return this;
    },
    // Add an item to the asynchronous iterable.
    add(x) {
      if (!waiting) return queued.push(Promise.resolve(x));

      waiting.resolve(x);
      waiting = null;
    },
    next() {
      if (queued.length) return {done: false, value: queued.shift()};
      if (done) return {done};

      waiting = deferred();
      const value = waiting.promise;

      return {done: false, value};
    },
    done() {
      done = true;
    },
  };
}

// Stretch out an asynchronous observable to emit no more often than specified.
async function* spaceOut(it, ms) {
  try {
    for await (const x of it) {
      yield x;
      await wait(ms);
    }
  } catch (e) {
    console.error("problem", e);
  }
}

async function test() {
  const it = asyncGeneratorOf();

  it.add(3);

  (async () => {
    for await (const x of spaceOut(it, 500)) console.log(x);
  })();

  it.add(4);

  // We can also add a promise.
  it.add(new Promise(resolve => setTimeout(() => resolve(49), 2000)));

  // Wait before adding the next item.
  await wait(5000);
  it.add(5);

  it.done();
}

test();