stream: allow stream to stay open after take

lib/internal/streams/operators.js

@@ -15,6 +15,7 @@ const {
   validateAbortSignal,
   validateInteger,
   validateObject,
+  validateBoolean,
 } = require('internal/validators');
 const { kWeakHandler } = require('internal/event_target');
 const { finished } = require('internal/streams/end-of-stream');
@@ -397,13 +398,16 @@ function take(number, options = undefined) {
   if (options?.signal != null) {
     validateAbortSignal(options.signal, 'options.signal');
   }
+  if (options?.destroyStream != null) {
+    validateBoolean(options.destroyStream, 'options.destroyStream');
+  }
 
   number = toIntegerOrInfinity(number);
   return async function* take() {
     if (options?.signal?.aborted) {
       throw new AbortError();
     }
-    for await (const val of this) {
+    for await (const val of this.iterator({ destroyOnReturn: options.destroyStream ?? true })) {
       if (options?.signal?.aborted) {
         throw new AbortError();
       }