Implement R2 multipart upload bindings (#486)

* Implement R2 multipart upload bindings

* fixup! Implement R2 multipart upload bindings

packages/core/src/standards/http.ts

@@ -158,6 +158,7 @@ export function _headersFromIncomingRequest(
 export const _kInner = Symbol("kInner");
 
 const kBodyStream = Symbol("kBodyStream");
+const kBodyStreamBrand = Symbol("kBodyStreamBrand");
 const kInputGated = Symbol("kInputGated");
 const kFormDataFiles = Symbol("kFormDataFiles");
 const kCloned = Symbol("kCloned");
@@ -197,7 +198,7 @@ export class Body<Inner extends BaseRequest | BaseResponse> {
 
  if (body === null) return body;
  // Only transform body stream once
- const bodyStream = this[kBodyStream];
+ let bodyStream = this[kBodyStream];
  if (bodyStream) return bodyStream;
  assert(body instanceof ReadableStream);
 
@@ -263,7 +264,9 @@ export class Body<Inner extends BaseRequest | BaseResponse> {
  cancel: (reason) => reader.cancel(reason),
  };
  // TODO: maybe set { highWaterMark: 0 } as a strategy here?
- return (this[kBodyStream] = new ReadableStream(source));
+ bodyStream = new ReadableStream(source);
+ Object.defineProperty(bodyStream, kBodyStreamBrand, { value: true });
+ return (this[kBodyStream] = bodyStream);
  }
  get bodyUsed(): boolean {
  return this[_kInner].bodyUsed;
@@ -357,6 +360,13 @@ export function withStringFormDataFiles<
  return body;
 }
 
+/** @internal */
+export function _isBodyStream(stream: ReadableStream): boolean {
+ return (
+ (stream as { [kBodyStreamBrand]?: unknown })[kBodyStreamBrand] === true
+ );
+}
+
 export type RequestInfo = BaseRequestInfo | Request;
 
 export interface RequestInit extends BaseRequestInit {

packages/core/src/standards/index.ts

@@ -10,6 +10,7 @@ export {
  Body,
  withInputGating,
  withStringFormDataFiles,
+ _isBodyStream,
  Request,
  withImmutableHeaders,
  Response,
@@ -48,6 +49,8 @@ export {
  CompressionStream,
  DecompressionStream,
  _isByteStream,
+ _isDisturbedStream,
+ _isFixedLengthStream,
 } from "./streams";
 export type { ArrayBufferViewConstructor } from "./streams";
 export * from "./navigator";

packages/core/src/standards/streams.ts

@@ -34,6 +34,19 @@ export function _isByteStream(
  return false;
 }
 
+/** @internal */
+export function _isDisturbedStream(stream: ReadableStream): boolean {
+ // Try to determine if stream is a stream has been consumed at all by
+ // inspecting its state.
+ for (const symbol of Object.getOwnPropertySymbols(stream)) {
+ if (symbol.description === "kState") {
+ // @ts-expect-error symbol properties are not included in type definitions
+ return !!stream[symbol].disturbed;
+ }
+ }
+ return false;
+}
+
 function convertToByteStream(
  stream: ReadableStream<Uint8Array>,
  clone = false
@@ -250,6 +263,13 @@ export class FixedLengthStream extends IdentityTransformStream {
  };
 }
 
+/** @internal */
+export function _isFixedLengthStream(stream: ReadableStream): boolean {
+ return (
+ (stream as { [kContentLength]?: unknown })[kContentLength] !== undefined
+ );
+}
+
 function createTransformerFromTransform(transform: Transform): Transformer {
  // TODO: backpressure? see https://github.com/nodejs/node/blob/440d95a878a1a19bf72a2685fc8fc0f47100b510/lib/internal/webstreams/adapters.js#L538
  return {

packages/core/test/standards/streams.spec.ts

@@ -14,6 +14,7 @@ import {
  Request,
  Response,
  _isByteStream,
+ _isDisturbedStream,
 } from "@miniflare/core";
 import { utf8Decode, utf8Encode } from "@miniflare/shared-test";
 import test, { Macro, ThrowsExpectation } from "ava";
@@ -74,6 +75,35 @@ test("_isByteStream: determines if a ReadableStream is a byte stream", (t) => {
  t.false(_isByteStream(regularStream));
  t.true(_isByteStream(byteStream));
 });
+test("_isDisturbedStream: determines if a ReadableStream is disturbed", async (t) => {
+ const regularStream = new ReadableStream({
+ pull(controller) {
+ controller.enqueue(new Uint8Array([1, 2, 3]));
+ controller.enqueue("thing");
+ controller.close();
+ },
+ });
+ const byteStream = new ReadableStream({
+ type: "bytes",
+ pull(controller) {
+ controller.enqueue(new Uint8Array([1, 2, 3]));
+ controller.close();
+ },
+ });
+ t.false(_isDisturbedStream(regularStream));
+ t.false(_isDisturbedStream(byteStream));
+
+ // @ts-expect-error ReadableStream types are incompatible
+ await arrayBuffer(regularStream);
+ t.true(_isDisturbedStream(regularStream));
+
+ const reader = byteStream.getReader();
+ t.false(_isDisturbedStream(byteStream));
+ await reader.read();
+ t.true(_isDisturbedStream(byteStream));
+ await reader.releaseLock();
+ t.true(_isDisturbedStream(byteStream));
+});
 
 test("ReadableStreamBYOBReader: readAtLeast: reads at least n bytes", async (t) => {
  const stream = chunkedStream([[1, 2, 3], [4], [5, 6]]);