Merge branch 'master' into je-yield

Author: trxcllnt

src/asynciterable/_sleep.ts

@@ -1,7 +1,7 @@
 import { AbortError } from '../aborterror.js';
 
-export function sleep(dueTime: number, signal?: AbortSignal) {
-  return new Promise<void>((resolve, reject) => {
+export function sleep(dueTime: number, signal?: AbortSignal, unref = false): Promise<void> {
+  return new Promise((resolve, reject) => {
     if (signal && signal.aborted) {
       reject(new AbortError());
     }
@@ -18,6 +18,10 @@ export function sleep(dueTime: number, signal?: AbortSignal) {
       resolve();
     }, dueTime);
 
+    if (unref && typeof id['unref'] === 'function') {
+      id['unref']();
+    }
+
     if (signal) {
       signal.addEventListener('abort', onAbort, { once: true });
     }

src/asynciterable/interval.ts

@@ -4,18 +4,20 @@ import { throwIfAborted } from '../aborterror.js';
 
 class IntervalAsyncIterable extends AsyncIterableX<number> {
   private _dueTime: number;
+  private _unref: boolean;
 
-  constructor(dueTime: number) {
+  constructor(dueTime: number, unref: boolean) {
     super();
     this._dueTime = dueTime;
+    this._unref = unref;
   }
 
   async *[Symbol.asyncIterator](signal?: AbortSignal) {
     throwIfAborted(signal);
 
     let i = 0;
     while (1) {
-      await sleep(this._dueTime, signal);
+      await sleep(this._dueTime, signal, this._unref);
       yield i++;
     }
   }
@@ -25,8 +27,9 @@ class IntervalAsyncIterable extends AsyncIterableX<number> {
  * Produces a new item in an async-iterable at the given interval cycle time.
  *
  * @param {number} dueTime The due time in milliseconds to spawn a new item.
+ * @param {boolean} [unref=false] Whether to unref the interval timer.
  * @returns {AsyncIterableX<number>} An async-iterable producing values at the specified interval.
  */
-export function interval(dueTime: number): AsyncIterableX<number> {
-  return new IntervalAsyncIterable(dueTime);
+export function interval(dueTime: number, unref = false): AsyncIterableX<number> {
+  return new IntervalAsyncIterable(dueTime, unref);
 }

src/asynciterable/operators/buffercountortime.ts

@@ -18,8 +18,7 @@ class BufferCountOrTime<TSource> extends AsyncIterableX<TSource[]> {
 
   async *[Symbol.asyncIterator](signal?: AbortSignal) {
     const buffer: TSource[] = [];
-
-    const timer = interval(this.maxWaitTime).pipe(map(() => timerEvent));
+    const timer = interval(this.maxWaitTime, true).pipe(map(() => timerEvent));
     const source = concat(this.source, of(ended));
     const merged = merge(source, timer);
 

src/asynciterable/operators/timeout.ts

@@ -62,7 +62,7 @@ export class TimeoutAsyncIterable<TSource> extends AsyncIterableX<TSource> {
           it.next().then((val) => {
             return { type: VALUE_TYPE, value: val };
           }),
-          sleep(this._dueTime, signal).then(() => {
+          sleep(this._dueTime, signal, true).then(() => {
             return { type: ERROR_TYPE };
           }),
         ]);