timers: introduce setInterval async iterator

doc/api/timers.md

@@ -363,6 +363,19 @@ added: v15.0.0
   * `signal` {AbortSignal} An optional `AbortSignal` that can be used to
     cancel the scheduled `Immediate`.
 
+### `timersPromises.setInterval([delay[, value[, options]]])`
+
+* `delay` {number} The number of milliseconds to wait between iterations.
+   **Default**: `1`.
+* `value` {any} A value with which the iterator returns.
+* `options` {Object}
+  * `ref` {boolean} Set to `false` to indicate that the scheduled `Timeout`
+    between iterations should not require the Node.js event loop to
+    remain active.
+    **Default**: `true`.
+  * `signal` {AbortSignal} An optional `AbortSignal` that can be used to
+    cancel the scheduled `Timeout` between operations.
+
 [Event Loop]: https://nodejs.org/en/docs/guides/event-loop-timers-and-nexttick/#setimmediate-vs-settimeout
 [`AbortController`]: globals.md#globals_class_abortcontroller
 [`TypeError`]: errors.md#errors_class_typeerror

lib/timers/promises.js

@@ -1,9 +1,11 @@
 'use strict';
 
 const {
+  Symbol,
   FunctionPrototypeBind,
   Promise,
   PromisePrototypeFinally,
+  PromiseResolve,
   PromiseReject,
 } = primordials;
 
@@ -15,7 +17,7 @@ const {
 
 const {
   AbortError,
-  codes: { ERR_INVALID_ARG_TYPE }
+  codes: { ERR_INVALID_ARG_TYPE, ERR_INVALID_ARG_VALUE }
 } = require('internal/errors');
 
 function cancelListenerHandler(clear, reject) {
@@ -125,7 +127,170 @@ function setImmediate(value, options = {}) {
       () => signal.removeEventListener('abort', oncancel)) : ret;
 }
 
+function setInterval(after, value, options = {}) {
+  const args = value !== undefined ? [value] : value;
+  if (options == null || typeof options !== 'object') {
+    throw new ERR_INVALID_ARG_TYPE(
+      'options',
+      'Object',
+      options);
+  }
+  const {
+    signal,
+    ref = true,
+    // Defers start of timers until the first iteration
+    wait = false,
+    // This has each invocation of iterator.next set up a new timeout
+    timeout: asTimeout = false,
+    // Skips intervals that are missed
+    skip = false
+    // Clears entire queue of callbacks when skip = true and the callbacks well missed the timeout
+  } = options;
+  if (signal !== undefined &&
+    (signal === null ||
+      typeof signal !== 'object' ||
+      !('aborted' in signal))) {
+    throw new ERR_INVALID_ARG_TYPE(
+      'options.signal',
+      'AbortSignal',
+      signal);
+  }
+  if (typeof ref !== 'boolean') {
+    throw new ERR_INVALID_ARG_TYPE(
+      'options.ref',
+      'boolean',
+      ref);
+  }
+  return {
+    [Symbol.asyncIterator]() {
+      // asTimeout can't skip as they always have intervals between each iteration
+      const resolveEarlyEnabled = !asTimeout && skip;
+      let timeout,
+        callbacks = [],
+        active = true,
+        missed = 0;
+
+      setIntervalCycle();
+
+      const iterator = {
+        async next() {
+          if (!active) {
+            return {
+              done: true,
+              value: undefined
+            };
+          }
+          // TODO(@jasnell): If a decision is made that this cannot be backported
+          // to 12.x, then this can be converted to use optional chaining to
+          // simplify the check.
+          if (signal && signal.aborted) {
+            return PromiseReject(new AbortError());
+          }
+          return new Promise(
+            (resolve, reject) => {
+              callbacks.push({ resolve, reject });
+              if (missed > 0) {
+                resolveNext();
+              }
+              setIntervalCycle();
+            }
+          );
+        },
+        async return() {
+          active = false;
+          clear();
+          resolveAll({
+            done: true,
+            value: undefined
+          });
+          if (signal) {
+            signal.removeEventListener('abort', oncancel);
+          }
+          return {
+            done: true,
+            value: undefined
+          };
+        }
+      };
+      if (signal) {
+        signal.addEventListener('abort', oncancel, { once: true });
+      }
+      return iterator;
+
+      function setIntervalCycle() {
+        if (!active) {
+          return;
+        }
+        if (timeout) {
+          return;
+        }
+        // Wait and asTimeout both imply a callback is required before setting up a timeout
+        if (!callbacks.length && (wait || asTimeout)) {
+          return;
+        }
+        missed = 0;
+        const currentTimeout = timeout = new Timeout(() => {
+          if (asTimeout && currentTimeout === timeout) {
+            // No need to clear here as we set to not repeat for asTimeout
+            timeout = undefined;
+          }
+          resolveNext();
+        }, after, undefined, !asTimeout, true);
+        if (!ref) timeout.unref();
+        insert(timeout, timeout._idleTimeout);
+      }
+
+      function resolveNext() {
+        if (!callbacks.length) {
+          if (resolveEarlyEnabled) {
+            missed += 1;
+          }
+          return;
+        }
+        const deferred = callbacks.shift();
+        if (deferred) {
+          const { resolve } = deferred;
+          resolve({
+            done: false,
+            value
+          });
+          missed -= 1;
+        }
+        if (missed > 0 && callbacks.length) {
+          // Loop till we have completed each missed interval that we have a callback for
+          resolveNext();
+        }
+      }
+
+      function resolveAll(value) {
+        callbacks.forEach(({ resolve }) => resolve(value));
+        callbacks = [];
+      }
+
+      function rejectAll(error) {
+        callbacks.forEach(({ reject }) => reject(error));
+        callbacks = [];
+      }
+
+      function clear() {
+        if (timeout) {
+          // eslint-disable-next-line no-undef
+          clearTimeout(timeout);
+          timeout = undefined;
+        }
+      }
+
+      function oncancel() {
+        clear();
+        rejectAll(new AbortError());
+      }
+
+    }
+  };
+}
+
 module.exports = {
   setTimeout,
   setImmediate,
+  setInterval,
 };

test/parallel/test-timers-promisified.js

@@ -15,10 +15,12 @@ const timerPromises = require('timers/promises');
 
 const setTimeout = promisify(timers.setTimeout);
 const setImmediate = promisify(timers.setImmediate);
+const setInterval = promisify(timers.setInterval);
 const exec = promisify(child_process.exec);
 
 assert.strictEqual(setTimeout, timerPromises.setTimeout);
 assert.strictEqual(setImmediate, timerPromises.setImmediate);
+assert.strictEqual(setInterval, timerPromises.setInterval);
 
 process.on('multipleResolves', common.mustNotCall());
 
@@ -50,6 +52,50 @@ process.on('multipleResolves', common.mustNotCall());
   }));
 }
 
+{
+  const iterable = setInterval(1, 'foobar');
+  const iterator = iterable[Symbol.asyncIterator]();
+  const promise = iterator.next();
+  promise
+    .then(common.mustCall((result) => {
+      assert.ok(!result.done);
+      assert.strictEqual(result.value, 'foobar');
+      return iterator.return();
+    }))
+    .then(common.mustCall());
+}
+
+{
+  const iterable = setInterval(1);
+  const iterator = iterable[Symbol.asyncIterator]();
+  const promise = iterator.next();
+  promise
+    .then(common.mustCall((result) => {
+      assert.ok(!result.done);
+      assert.strictEqual(result.value, undefined);
+      return iterator.return();
+    }))
+    .then(common.mustCall());
+}
+
+{
+  const iterable = setInterval(1, 'foobar');
+  const iterator = iterable[Symbol.asyncIterator]();
+  const promise = iterator.next();
+  promise
+    .then(common.mustCall((result) => {
+      assert.ok(!result.done);
+      assert.strictEqual(result.value, 'foobar');
+      return iterator.next();
+    }))
+    .then(common.mustCall((result) => {
+      assert.ok(!result.done);
+      assert.strictEqual(result.value, 'foobar');
+      return iterator.return();
+    }))
+    .then(common.mustCall());
+}
+
 {
   const ac = new AbortController();
   const signal = ac.signal;
@@ -78,6 +124,33 @@ process.on('multipleResolves', common.mustNotCall());
   assert.rejects(setImmediate(10, { signal }), /AbortError/);
 }
 
+{
+  const ac = new AbortController();
+  const signal = ac.signal;
+  ac.abort(); // Abort in advance
+
+  const iterable = setInterval(1, undefined, { signal });
+  const iterator = iterable[Symbol.asyncIterator]();
+
+  assert.rejects(iterator.next(), /AbortError/);
+}
+
+{
+  const ac = new AbortController();
+  const signal = ac.signal;
+
+  const iterable = setInterval(100, undefined, { signal });
+  const iterator = iterable[Symbol.asyncIterator]();
+
+  // This promise should take 100 seconds to resolve, so now aborting it should
+  // mean we abort early
+  const promise = iterator.next();
+
+  ac.abort(); // Abort in after we have a next promise
+
+  assert.rejects(promise, /AbortError/);
+}
+
 {
   // Check that aborting after resolve will not reject.
   const ac = new AbortController();
@@ -148,6 +221,10 @@ process.on('multipleResolves', common.mustNotCall());
       (ref) => assert.rejects(setTimeout(10, null, { ref })), {
         code: 'ERR_INVALID_ARG_TYPE'
       })).then(common.mustCall());
+
+  [1, '', Infinity, null, {}].forEach((ref) => {
+    assert.throws(() => setInterval(10, undefined, { ref }));
+  });
 }
 
 {
@@ -160,8 +237,52 @@ process.on('multipleResolves', common.mustNotCall());
 
 {
   exec(`${process.execPath} -pe "const assert = require('assert');` +
-       'require(\'timers/promises\').setImmediate(null, { ref: false }).' +
-       'then(assert.fail)"').then(common.mustCall(({ stderr }) => {
+    'require(\'timers/promises\').setImmediate(null, { ref: false }).' +
+    'then(assert.fail)"').then(common.mustCall(({ stderr }) => {
+    assert.strictEqual(stderr, '');
+  }));
+}
+
+{
+  exec(`${process.execPath} -pe "const assert = require('assert');` +
+    'const interval = require(\'timers/promises\')' +
+    '.setInterval(1000, null, { ref: false });' +
+    'interval[Symbol.asyncIterator]().next()' +
+    '.then(assert.fail)"').then(common.mustCall(({ stderr }) => {
     assert.strictEqual(stderr, '');
   }));
 }
+
+{
+  const ac = new AbortController();
+  const input = 'foobar';
+  const signal = ac.signal;
+
+  const mainInterval = 5;
+  const loopInterval = mainInterval * 1.5;
+
+  const interval = setInterval(mainInterval, input, { signal, timeout: true });
+
+  async function runInterval(fn) {
+    const times = [];
+    for await (const value of interval) {
+      const index = times.length;
+      times[index] = [Date.now()];
+      assert.strictEqual(value, input);
+      await fn();
+      times[index] = [...times[index], Date.now()];
+    }
+  }
+
+  const noopLoop = runInterval(() => {});
+  const timeoutLoop = runInterval(() => setTimeout(loopInterval));
+
+  // Let it loop 5 times, then abort before the next
+  setTimeout(Math.floor(loopInterval * 5.5), undefined, { timeout: true }).then(common.mustCall(() => {
+    ac.abort();
+  }));
+
+  assert.rejects(noopLoop, /AbortError/);
+  assert.rejects(timeoutLoop, /AbortError/);
+
+}