stream: accept ArrayBuffer in CompressionStream and DecompressionStream

PR-URL: #61913Fixes: #43433
Reviewed-By: Mattias Buelens <mattias@buelens.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Jason Zhang <xzha4350@gmail.com>
Reviewed-By: René <contact.9a5d6388@renegade334.me.uk>
Reviewed-By: Anna Henningsen <anna@addaleax.net>

Author: suuuuuuminnnnnn

lib/internal/webstreams/adapters.js

@@ -53,6 +53,10 @@ const {
  Buffer,
 }=require('buffer');
 
+const{
+ isArrayBuffer,
+}=require('internal/util/types');
+
 const{
  AbortError,
  ErrnoException,
@@ -213,6 +217,9 @@ function newWritableStreamFromStreamWritable(streamWritable) {
 start(c){controller=c;},
 
 write(chunk){
+if(!streamWritable.writableObjectMode&&isArrayBuffer(chunk)){
+chunk=newUint8Array(chunk);
+}
 if(streamWritable.writableNeedDrain||!streamWritable.write(chunk)){
 backpressurePromise=PromiseWithResolvers();
 returnSafePromisePrototypeFinally(

test/parallel/test-webstreams-compression-buffer-source.js

@@ -0,0 +1,42 @@
+'use strict';
+require('../common');
+constassert=require('assert');
+consttest=require('node:test');
+const{ DecompressionStream, CompressionStream }=require('stream/web');
+
+// Minimal gzip-compressed bytes for "hello"
+constcompressedGzip=newUint8Array([
+31,139,8,0,0,0,0,0,0,3,
+203,72,205,201,201,7,0,134,166,16,54,5,0,0,0,
+]);
+
+test('DecompressionStream accepts ArrayBuffer chunks',async()=>{
+constds=newDecompressionStream('gzip');
+constwriter=ds.writable.getWriter();
+
+constwritePromise=writer.write(compressedGzip.buffer);
+writer.close();
+
+constchunks=awaitArray.fromAsync(ds.readable);
+awaitwritePromise;
+constout=Buffer.concat(chunks.map((c)=>Buffer.from(c)));
+assert.strictEqual(out.toString(),'hello');
+});
+
+test('CompressionStream round-trip with ArrayBuffer input',async()=>{
+constcs=newCompressionStream('gzip');
+constds=newDecompressionStream('gzip');
+
+constcsWriter=cs.writable.getWriter();
+
+constinput=newTextEncoder().encode('hello').buffer;
+
+awaitcsWriter.write(input);
+csWriter.close();
+
+awaitcs.readable.pipeTo(ds.writable);
+
+constout=awaitArray.fromAsync(ds.readable);
+constresult=Buffer.concat(out.map((c)=>Buffer.from(c)));
+assert.strictEqual(result.toString(),'hello');
+});

test/parallel/test-whatwg-webstreams-adapters-to-writablestream.js

@@ -165,3 +165,35 @@ class TestWritable extends Writable {
 constwriter=writableStream.getWriter();
 writer.closed.then(common.mustCall());
 }
+
+{
+constduplex=newPassThrough();
+constwritableStream=newWritableStreamFromStreamWritable(duplex);
+constec=newTextEncoder();
+constarrayBuffer=ec.encode('hello').buffer;
+writableStream
+.getWriter()
+.write(arrayBuffer)
+.then(common.mustCall());
+
+duplex.on('data',common.mustCall((chunk)=>{
+assert(chunkinstanceofBuffer);
+assert(chunk.equals(Buffer.from('hello')));
+}));
+}
+
+{
+constduplex=newPassThrough({objectMode: true});
+constwritableStream=newWritableStreamFromStreamWritable(duplex);
+constec=newTextEncoder();
+constarrayBuffer=ec.encode('hello').buffer;
+writableStream
+.getWriter()
+.write(arrayBuffer)
+.then(common.mustCall());
+
+duplex.on('data',common.mustCall((chunk)=>{
+assert(chunkinstanceofArrayBuffer);
+assert.strictEqual(chunk,arrayBuffer);
+}));
+}

test/wpt/status/compression.json

@@ -5,9 +5,6 @@
 "decompression-bad-chunks.tentative.any.js": {
 "skip": "Execution \"hangs\", ArrayBuffer and TypedArray is not accepted and throws, instead of rejects during writer.write"
  },
-"decompression-buffersource.tentative.any.js": {
-"skip": "ArrayBuffer and TypedArray is not accepted and throws, instead of rejects during writer.write"
- },
 "compression-with-detach.tentative.window.js": {
 "requires": ["crypto"]
  },