stream: expose ReadableStreamTee

Signed-off-by: Matteo Collina <hello@matteocollina.com>
PR-URL: #64195
Refs: nodejs/undici#5358
Reviewed-By: Matthew Aitken <maitken033380023@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: mcollina

doc/api/webstreams.md

@@ -106,6 +106,28 @@ For more details refer to the relevant documentation:
 
 ## API
 
+### `ReadableStreamTee(stream[, cloneForBranch2])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+*`stream` {ReadableStream}
+*`cloneForBranch2` {boolean} When `true`, chunks enqueued into the second
+ branch are cloned from chunks enqueued into the first branch. **Default:**
+`false`.
+* Returns: {ReadableStream\[]} Two {ReadableStream} branches.
+
+Runs the WHATWG `ReadableStreamTee` abstract operation on `stream`.
+
+This differs from `readableStream.tee()` only when `cloneForBranch2` is
+`true`. The `tee()` method always passes `false`, while other web platform
+specifications, such as Fetch body cloning, pass `true` so that the second
+branch receives cloned chunks and consumption of one branch cannot mutate chunks
+seen by the other.
+
 ### Class: `ReadableStream`
 
 <!-- YAML

lib/stream/web.js

@@ -18,8 +18,20 @@ const {
  ReadableStreamBYOBRequest,
  ReadableByteStreamController,
  ReadableStreamDefaultController,
+ isReadableStream,
+ readableStreamTee,
 }=require('internal/webstreams/readablestream');
 
+const{
+codes: {
+ERR_INVALID_ARG_TYPE,
+},
+}=require('internal/errors');
+
+const{
+ validateBoolean,
+}=require('internal/validators');
+
 const{
  ByteLengthQueuingStrategy,
  CountQueuingStrategy,
@@ -35,8 +47,17 @@ const {
  DecompressionStream,
 }=require('internal/webstreams/compression');
 
+functionReadableStreamTee(stream,cloneForBranch2=false){
+if(!isReadableStream(stream)){
+thrownewERR_INVALID_ARG_TYPE('stream','ReadableStream',stream);
+}
+validateBoolean(cloneForBranch2,'cloneForBranch2');
+returnreadableStreamTee(stream,cloneForBranch2);
+}
+
 module.exports={
  ReadableStream,
+ ReadableStreamTee,
  ReadableStreamDefaultReader,
  ReadableStreamBYOBReader,
  ReadableStreamBYOBRequest,

test/parallel/test-whatwg-readablestream.js

@@ -15,6 +15,7 @@ const {
  ByteLengthQueuingStrategy,
  CountQueuingStrategy,
  ReadableStream,
+ ReadableStreamTee,
  ReadableStreamDefaultReader,
  ReadableStreamDefaultController,
  ReadableByteStreamController,
@@ -1527,6 +1528,28 @@ class Source {
 }).then(common.mustCall());
 }
 
+{
+// Test public ReadableStreamTee() cloneForBranch2 argument
+assert.strictEqual(typeofReadableStreamTee,'function');
+constchunk=newUint8Array([65]);
+constreadable=newReadableStream({
+start(controller){
+controller.enqueue(chunk);
+controller.close();
+},
+});
+const[r1,r2]=ReadableStreamTee(readable,true);
+
+(async()=>{
+const{value: value1}=awaitr1.getReader().read();
+assert.strictEqual(value1[0],65);
+value1[0]=66;
+
+const{value: value2}=awaitr2.getReader().read();
+assert.strictEqual(value2[0],65);
+})().then(common.mustCall());
+}
+
 {
 // Test tee() cloneForBranch2 argument
 constreadable=newReadableStream({