stream: unref

lib/internal/streams/readable.js

@@ -33,6 +33,7 @@ const {
   SafeSet,
   SymbolAsyncIterator,
   Symbol,
+  ReflectApply,
 } = primordials;
 
 module.exports = Readable;
@@ -1154,6 +1155,39 @@ async function* createAsyncIterator(stream, options) {
   }
 }
 
+function staticUnref(stream) {
+  // TODO - We can probably have a better way to do this
+  function unref() {
+    return async function* unref() {
+      const iterator = this.iterator({ destroyOnReturn: false });
+
+      for await (const val of iterator) {
+        yield val;
+      }
+    }.call(this);
+  }
+
+  // Instead of regular operator registration
+  const newStream = Readable.from(ReflectApply(unref, stream, []));
+
+  // TODO - we should have better way instead of listening to those events, no?
+  // stream.once('close', () => {
+  //   destroyImpl.destroyer(newStream, null);
+  // });
+  stream.once('error', (e) => {
+    destroyImpl.destroyer(newStream, e);
+  });
+
+  newStream.on('error', (e) => {
+    if (e.name !== 'AbortError') {
+      destroyImpl.destroyer(stream, e);
+    }
+  });
+
+  return newStream;
+}
+
+
 // Making it explicit these properties are not enumerable
 // because otherwise some prototype manipulation in
 // userland will fail.
@@ -1398,6 +1432,10 @@ Readable.from = function(iterable, opts) {
   return from(Readable, iterable, opts);
 };
 
+Readable.unref = function(stream) {
+  return staticUnref(stream);
+};
+
 let webStreamsAdapters;
 
 // Lazy to avoid circular references

test/parallel/test-stream-unref.js

@@ -0,0 +1,95 @@
+'use strict';
+require('../common');
+
+const {
+  Readable,
+} = require('stream');
+const { it } = require('node:test');
+const { strictEqual } = require('assert');
+
+const { from, unref } = Readable;
+
+const nextTick = () => new Promise((resolve) => process.nextTick(resolve));
+
+it('unref stream error should propagate to original one', async () => {
+  const originalStream = from([1, 2, 3, 4, 5]);
+
+  let emittedError;
+  originalStream.on('error', (e) => {
+    emittedError = e;
+  });
+  const unrefStream = unref(originalStream);
+
+  const thrownError = new Error('test');
+
+  unrefStream.destroy(thrownError);
+
+  await nextTick();
+  strictEqual(unrefStream.destroyed, true);
+  strictEqual(originalStream.destroyed, true);
+
+  // Need another tick to propagate the error
+  await nextTick();
+  strictEqual(emittedError, thrownError);
+});
+
+it('Original stream error should propagate to unref one', async () => {
+  const originalStream = from([1, 2, 3, 4, 5]);
+  const unrefStream = unref(originalStream);
+
+  let emittedError;
+  unrefStream.on('error', (e) => {
+    emittedError = e;
+  });
+
+  const thrownError = new Error('test');
+
+  originalStream.destroy(thrownError);
+
+  await nextTick();
+  strictEqual(unrefStream.destroyed, true);
+  strictEqual(originalStream.destroyed, true);
+
+  await nextTick();
+  strictEqual(emittedError, thrownError);
+});
+
+it('Should not close original stream when unref one finished but not consumed all data', async () => {
+  const originalStream = from([1, 2, 3, 4, 5]);
+
+  const unrefStream = unref(originalStream);
+
+  // eslint-disable-next-line no-unused-vars
+  for await (const _ of unrefStream) {
+    break;
+  }
+
+  await nextTick();
+  strictEqual(unrefStream.destroyed, true);
+  strictEqual(originalStream.destroyed, false);
+});
+
+it('Should close original stream when unref one consume all data', async () => {
+  const originalStream = from([1, 2, 3, 4, 5]);
+
+  const unrefStream = unref(originalStream);
+
+  // This throw an abort error
+  await unrefStream.toArray();
+
+  await nextTick();
+
+  strictEqual(unrefStream.destroyed, true);
+  strictEqual(originalStream.destroyed, true);
+});
+
+// This fail
+it('original stream close should close unref one', async () => {
+  const originalStream = from([1, 2, 3, 4, 5]);
+  const unrefStream = unref(originalStream);
+
+  await originalStream.toArray();
+
+  strictEqual(originalStream.destroyed, true);
+  strictEqual(unrefStream.destroyed, true);
+});