Merge pull request #5428 from cloudflare/jasnell/replace-bodyinputstream

Author: jasnell

src/workerd/api/http.c++

@@ -591,53 +591,6 @@ jsg::Ref<Headers> Headers::deserialize(
 }
 
 // =======================================================================================
-
-namespace {
-
-class BodyBufferInputStream final: public ReadableStreamSource {
- public:
-  BodyBufferInputStream(Body::Buffer buffer)
-      : unread(buffer.view),
-        ownBytes(kj::mv(buffer.ownBytes)) {}
-
-  kj::Promise<size_t> tryRead(void* buffer, size_t minBytes, size_t maxBytes) override {
-    if (unread != nullptr) {
-      size_t amount = kj::min(maxBytes, unread.size());
-      memcpy(buffer, unread.begin(), amount);
-      unread = unread.slice(amount, unread.size());
-      return amount;
-    }
-
-    return static_cast<size_t>(0);
-  }
-
-  kj::Maybe<uint64_t> tryGetLength(StreamEncoding encoding) override {
-    if (encoding == StreamEncoding::IDENTITY) {
-      return unread.size();
-    } else {
-      // Who knows what the compressed size will be?
-      return kj::none;
-    }
-  }
-
-  kj::Promise<DeferredProxy<void>> pumpTo(WritableStreamSink& output, bool end) override {
-    if (unread != nullptr) {
-      auto data = unread;
-      unread = nullptr;
-      co_await output.write(data);
-      if (end) co_await output.end();
-    }
-
-    co_return;
-  }
-
- private:
-  kj::ArrayPtr<const byte> unread;
-  kj::OneOf<kj::Own<Body::RefcountedBytes>, jsg::Ref<Blob>> ownBytes;
-};
-
-}  // namespace
-
 // Make an array of characters containing random hexadecimal digits.
 //
 // Note: Rather than use random hex digits, we could generate the hex digits by hashing the
@@ -725,10 +678,14 @@ Body::ExtractedBody Body::extractBody(jsg::Lock& js, Initializer init) {
     }
   }
 
-  auto bodyStream = kj::heap<BodyBufferInputStream>(buffer.clone(js));
-
-  return {js.alloc<ReadableStream>(IoContext::current(), kj::mv(bodyStream)), kj::mv(buffer),
-    kj::mv(contentType)};
+  auto clonedBuffer = buffer.clone(js);
+  auto memStream = newMemoryInputStream(clonedBuffer.view, kj::heap(kj::mv(clonedBuffer.ownBytes)));
+  auto rs = newSystemStream(kj::mv(memStream), StreamEncoding::IDENTITY, IoContext::current());
+  return {
+    js.alloc<ReadableStream>(IoContext::current(), kj::mv(rs)),
+    kj::mv(buffer),
+    kj::mv(contentType),
+  };
 }
 
 Body::Body(jsg::Lock& js, kj::Maybe<ExtractedBody> init, Headers& headers)
@@ -775,8 +732,9 @@ void Body::rewindBody(jsg::Lock& js) {
 
   KJ_IF_SOME(i, impl) {
     auto bufferCopy = KJ_ASSERT_NONNULL(i.buffer).clone(js);
-    auto bodyStream = kj::heap<BodyBufferInputStream>(kj::mv(bufferCopy));
-    i.stream = js.alloc<ReadableStream>(IoContext::current(), kj::mv(bodyStream));
+    auto memStream = newMemoryInputStream(bufferCopy.view, kj::heap(kj::mv(bufferCopy.ownBytes)));
+    auto rs = newSystemStream(kj::mv(memStream), StreamEncoding::IDENTITY, IoContext::current());
+    i.stream = js.alloc<ReadableStream>(IoContext::current(), kj::mv(rs));
   }
 }