Merge pull request #5014 from cloudflare/jasnell/streams-refactor-next-gen

Author: jasnell

src/workerd/api/BUILD.bazel

@@ -347,6 +347,24 @@ kj_test(
     deps = ["//src/workerd/tests:test-fixture"],
 )
 
+kj_test(
+    src = "streams/writable-sink-adapter-test.c++",
+    deps = [
+        "//src/workerd/io",
+        "//src/workerd/jsg",
+        "//src/workerd/tests:test-fixture",
+    ],
+)
+
+kj_test(
+    src = "streams/readable-source-adapter-test.c++",
+    deps = [
+        "//src/workerd/io",
+        "//src/workerd/jsg",
+        "//src/workerd/tests:test-fixture",
+    ],
+)
+
 wd_test(
     src = "actor-alarms-delete-test.wd-test",
     args = ["--experimental"],

src/workerd/api/streams/common.h

@@ -593,7 +593,8 @@ class WritableStreamController {
     //
     // The controller is guaranteed to either outlive the Writer or will detach the Writer so the
     // WritableStreamController& reference should always remain valid.
-    virtual void attach(WritableStreamController& controller,
+    virtual void attach(jsg::Lock& js,
+        WritableStreamController& controller,
         jsg::Promise<void> closedPromise,
         jsg::Promise<void> readyPromise) = 0;
 
@@ -603,7 +604,7 @@ class WritableStreamController {
 
     // The ready promise can be replaced whenever backpressure is signaled by the underlying
     // controller.
-    virtual void replaceReadyPromise(jsg::Promise<void> readyPromise) = 0;
+    virtual void replaceReadyPromise(jsg::Lock& js, jsg::Promise<void> readyPromise) = 0;
   };
 
   struct PendingAbort {
@@ -821,10 +822,10 @@ class WriterLocked {
     return readyFulfiller;
   }
 
-  void setReadyFulfiller(jsg::PromiseResolverPair<void>& pair) {
+  void setReadyFulfiller(jsg::Lock& js, jsg::PromiseResolverPair<void>& pair) {
     KJ_IF_SOME(w, writer) {
       readyFulfiller = kj::mv(pair.resolver);
-      w.replaceReadyPromise(kj::mv(pair.promise));
+      w.replaceReadyPromise(js, kj::mv(pair.promise));
     }
   }
 

src/workerd/api/streams/internal.c++

@@ -1023,7 +1023,7 @@ void WritableStreamInternalController::updateBackpressure(jsg::Lock& js, bool ba
       // the existing one is resolved or not.
       auto prp = js.newPromiseAndResolver<void>();
       prp.promise.markAsHandled(js);
-      writerLock.setReadyFulfiller(prp);
+      writerLock.setReadyFulfiller(js, prp);
       return;
     }
 
@@ -1406,7 +1406,7 @@ bool WritableStreamInternalController::lockWriter(jsg::Lock& js, Writer& writer)
   }
 
   writeState = kj::mv(lock);
-  writer.attach(*this, kj::mv(closedPrp.promise), kj::mv(readyPrp.promise));
+  writer.attach(js, *this, kj::mv(closedPrp.promise), kj::mv(readyPrp.promise));
   return true;
 }