webstreams: native CompressionStream/DecompressionStream/TextEncoderStream/TextDecoderStream

src/codegen/generate-jssink.ts

@@ -169,7 +169,7 @@ function header() {
                 static size_t memoryCost(void* sinkPtr);
 
                 ${controller}(JSC::VM& vm, JSC::Structure* structure, void* sinkPtr, uintptr_t onDestroy)
-                    : Base(vm, structure, sinkPtr, onDestroy)
+                    : Base(vm, structure, sinkPtr, SinkID::${name}, onDestroy)
                 {
                 }
 
@@ -208,19 +208,24 @@ class JSReadableSinkControllerBase : public JSC::JSDestructibleObject {
 public:
     using Base = JSC::JSDestructibleObject;
 
+    DECLARE_INFO;
+
     void* wrapped() const { return m_sinkPtr; }
+    SinkID sinkId() const { return m_sinkId; }
 
     void* m_sinkPtr;
+    SinkID m_sinkId;
     mutable WriteBarrier<JSC::JSObject> m_onPull;
     mutable WriteBarrier<JSC::JSObject> m_onClose;
     mutable JSC::Weak<JSObject> m_weakReadableStream;
     uintptr_t m_onDestroy { 0 };
 
 protected:
-    JSReadableSinkControllerBase(JSC::VM& vm, JSC::Structure* structure, void* sinkPtr, uintptr_t onDestroy)
+    JSReadableSinkControllerBase(JSC::VM& vm, JSC::Structure* structure, void* sinkPtr, SinkID sinkId, uintptr_t onDestroy)
         : Base(vm, structure)
     {
         m_sinkPtr = sinkPtr;
+        m_sinkId = sinkId;
         m_onDestroy = onDestroy;
     }
 };
@@ -301,6 +306,8 @@ using namespace JSC;
 
 ${classes.map(name => `extern "C" size_t ${name}__memoryCost(void* sinkPtr);`).join("\n")}
 ${classes.map(name => `extern "C" void ${name}__controllerDetached(void* sinkPtr, JSC::EncodedJSValue controllerValue);`).join("\n")}
+
+const ClassInfo JSReadableSinkControllerBase::s_info = { "ReadableSinkController"_s, &Base::s_info, nullptr, nullptr, CREATE_METHOD_TABLE(JSReadableSinkControllerBase) };
 `;
   var templ = head;
 

src/event_loop/ConcurrentTask.rs

@@ -115,6 +115,7 @@ pub mod task_tag {
         NativeBrotli,
         NativeZlib,
         NativeZstd,
+        CompressionStreamCoderTask,
         Open,
         PasswordHashResult,
         PasswordVerifyResult,

src/js/builtins.d.ts

@@ -410,8 +410,6 @@ declare function $streamErrored(): TODO;
 declare function $streamReadable(): TODO;
 declare function $streamWritable(): TODO;
 declare function $syscall(): TODO;
-declare function $textDecoderStreamDecoder(): TODO;
-declare function $textEncoderStreamEncoder(): TODO;
 declare function $toNamespacedPath(): TODO;
 declare function $url(): TODO;
 declare function $view(): TODO;
@@ -627,7 +625,6 @@ declare function $ERR_STREAM_CANNOT_PIPE(): Error;
 declare function $ERR_STREAM_WRITE_AFTER_END(): Error;
 declare function $ERR_STREAM_UNSHIFT_AFTER_END_EVENT(): Error;
 declare function $ERR_STREAM_PUSH_AFTER_EOF(): Error;
-declare function $ERR_TRAILING_JUNK_AFTER_STREAM_END(): TypeError;
 declare function $ERR_STREAM_UNABLE_TO_PIPE(): Error;
 declare function $ERR_ILLEGAL_CONSTRUCTOR(): TypeError;
 declare function $ERR_SERVER_ALREADY_LISTEN(): Error;

src/js/builtins/BunBuiltinNames.h

@@ -33,7 +33,6 @@ using namespace JSC;
     macro(ReadableStreamDefaultController) \
     macro(ReadableStreamDefaultReader) \
     macro(SQL) \
-    macro(TextEncoderStreamEncoder) \
     macro(TransformStream) \
     macro(TransformStreamDefaultController) \
     macro(WritableStream) \
@@ -80,7 +79,6 @@ using namespace JSC;
     macro(disturbed) \
     macro(domain) \
     macro(drain) \
-    macro(encode) \
     macro(encoding) \
     macro(end) \
     macro(errno) \

src/js/internal/webstreams_adapters.ts

@@ -33,9 +33,6 @@ const SafePromisePrototypeFinally = $Promise.prototype.finally;
 
 const constants_zlib = $processBindingConstants.zlib;
 
-const kValidateChunk = Symbol("kValidateChunk");
-const kDestroyOnSyncError = Symbol("kDestroyOnSyncError");
-
 function tryTransferToNativeReadable(stream, options) {
   const ptr = stream.$bunNativePtr;
   if (!ptr || ptr === -1) {
@@ -191,7 +188,7 @@ function handleKnownInternalErrors(cause: Error | null): Error | null {
 
 const noop = () => {};
 
-function newWritableStreamFromStreamWritable(streamWritable, options = kEmptyObject) {
+function newWritableStreamFromStreamWritable(streamWritable) {
   // Not using the internal/streams/utils isWritableNodeStream utility
   // here because it will return false if streamWritable is a Duplex
   // whose writable option is false. For a Duplex that is not writable,
@@ -268,34 +265,20 @@ function newWritableStreamFromStreamWritable(streamWritable, options = kEmptyObj
       },
 
       write(chunk) {
-        try {
-          options[kValidateChunk]?.(chunk);
-          if (!streamWritable.writableObjectMode && isAnyArrayBuffer(chunk)) {
-            chunk = new Uint8Array(chunk);
-          }
-          const needDrainBefore = streamWritable.writableNeedDrain;
-          if (needDrainBefore || !streamWritable.write(chunk)) {
-            backpressurePromise = PromiseWithResolvers();
-            // write() may set writableNeedDrain; the post-write value is
-            // what decides whether we resolve immediately.
-            if (!streamWritable.writableNeedDrain) {
-              backpressurePromise.resolve();
-            }
-            return SafePromisePrototypeFinally.$call(backpressurePromise.promise, () => {
-              backpressurePromise = undefined;
-            });
-          }
-        } catch (error) {
-          // When the kDestroyOnSyncError flag is set (e.g. for
-          // CompressionStream), a sync throw must also destroy the
-          // stream so the readable side is errored too. Without this
-          // the readable side hangs forever. This replicates the
-          // TransformStream semantics: error both sides on any throw
-          // in the transform path.
-          if (options[kDestroyOnSyncError]) {
-            destroyer(streamWritable, error);
+        if (!streamWritable.writableObjectMode && isAnyArrayBuffer(chunk)) {
+          chunk = new Uint8Array(chunk);
+        }
+        const needDrainBefore = streamWritable.writableNeedDrain;
+        if (needDrainBefore || !streamWritable.write(chunk)) {
+          backpressurePromise = PromiseWithResolvers();
+          // write() may set writableNeedDrain; the post-write value is
+          // what decides whether we resolve immediately.
+          if (!streamWritable.writableNeedDrain) {
+            backpressurePromise.resolve();
           }
-          throw error;
+          return SafePromisePrototypeFinally.$call(backpressurePromise.promise, () => {
+            backpressurePromise = undefined;
+          });
         }
       },
 
@@ -634,15 +617,7 @@ function newReadableWritablePairFromDuplex(duplex, options = kEmptyObject) {
     return { readable, writable };
   }
 
-  const writableOptions = {
-    __proto__: null,
-    [kValidateChunk]: options[kValidateChunk],
-    [kDestroyOnSyncError]: options[kDestroyOnSyncError],
-  };
-
-  const writable = isWritable(duplex)
-    ? newWritableStreamFromStreamWritable(duplex, writableOptions)
-    : new WritableStream();
+  const writable = isWritable(duplex) ? newWritableStreamFromStreamWritable(duplex) : new WritableStream();
 
   if (!isWritable(duplex)) writable.close();
 
@@ -840,31 +815,12 @@ function newStreamDuplexFromReadableWritablePair(pair = kEmptyObject, options =
   return duplex;
 }
 
-// Shared by CompressionStream and DecompressionStream: per the Compression
-// Streams spec, chunks must be BufferSource (ArrayBuffer or ArrayBufferView
-// not backed by SharedArrayBuffer), and an invalid chunk must error both
-// sides of the pair synchronously.
-function newBufferSourceTransformPairFromDuplex(duplex) {
-  const { isArrayBufferView, isSharedArrayBuffer } = require("node:util/types");
-  return newReadableWritablePairFromDuplex(duplex, {
-    [kValidateChunk]: function validateBufferSourceChunk(chunk) {
-      if (isSharedArrayBuffer(isArrayBufferView(chunk) ? chunk.buffer : chunk)) {
-        throw $ERR_INVALID_ARG_TYPE("chunk", ["ArrayBuffer", "Buffer", "TypedArray", "DataView"], chunk);
-      }
-    },
-    [kDestroyOnSyncError]: true,
-  });
-}
-
 export default {
   newWritableStreamFromStreamWritable,
   newReadableStreamFromStreamReadable,
   newStreamWritableFromWritableStream,
   newStreamReadableFromReadableStream,
   newReadableWritablePairFromDuplex,
   newStreamDuplexFromReadableWritablePair,
-  newBufferSourceTransformPairFromDuplex,
-  kValidateChunk,
-  kDestroyOnSyncError,
   _ReadableFromWeb: ReadableFromWeb,
 };

src/js/node/zlib.ts

@@ -197,8 +197,6 @@ function ZlibBase(opts, mode, handle, { flush, finishFlush, fullFlush }) {
   this._defaultFullFlushFlag = fullFlush;
   this._info = opts && opts.info;
   this._maxOutputLength = maxOutputLength;
-
-  this._rejectGarbageAfterEnd = opts?.rejectGarbageAfterEnd === true;
 }
 $toClass(ZlibBase, "ZlibBase", Transform);
 
@@ -509,13 +507,6 @@ function processCallback() {
     // This applies to streams where we don't check data past the end of
     // what was consumed; that is, everything except Gunzip/Unzip.
 
-    if (self._rejectGarbageAfterEnd) {
-      const err = $ERR_TRAILING_JUNK_AFTER_STREAM_END();
-      self.destroy(err);
-      this.cb(err);
-      return;
-    }
-
     self.push(null);
   }