Support fetch api request stream

packages/connect-web-test/src/crosstest/client_streaming.spec.ts

@@ -0,0 +1,42 @@
+// Copyright 2021-2023 Buf Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import { createPromiseClient } from "@bufbuild/connect";
+import { Payload, PayloadType } from "../gen/grpc/testing/messages_pb.js";
+import { TestService } from "../gen/grpc/testing/test_connect.js";
+import { describeTransports } from "../helpers/crosstestserver.js";
+
+describe("client_streaming", function () {
+  describeTransports((transport) => {
+    const sizes = [31415, 9, 2653, 58979];
+    it("with promise client", async function () {
+      async function* input() {
+        for (const size of sizes) {
+          yield {
+            payload: new Payload({
+              body: new Uint8Array(size),
+              type: PayloadType.COMPRESSABLE,
+            }),
+          };
+        }
+        await new Promise((resolve) => setTimeout(resolve, 1));
+      }
+      const client = createPromiseClient(TestService, transport());
+      const { aggregatedPayloadSize } = await client.streamingInputCall(
+        input()
+      );
+      expect(aggregatedPayloadSize).toBe(sizes.reduce((p, c) => p + c, 0));
+    });
+  });
+});

packages/connect-web/src/connect-envelop-stream.ts

@@ -0,0 +1,16 @@
+import { encodeEnvelope } from "@bufbuild/connect/protocol";
+
+export class ConnectEnvelopStream<I> extends TransformStream {
+  public constructor(serializer: (data: I) => Uint8Array) {
+    super({
+      transform(chunk, controller) {
+        try {
+          const envelope = encodeEnvelope(0, serializer(chunk));
+          controller.enqueue(envelope);
+        } catch (e) {
+          controller.error(e);
+        }
+      },
+    });
+  }
+}

packages/connect-web/src/connect-transport.ts

@@ -50,6 +50,8 @@ import {
   validateResponse,
 } from "@bufbuild/connect/protocol-connect";
 import { assertFetchApi } from "./assert-fetch-api.js";
+import { supportsRequestStreams } from "./supports-request-streams.js";
+import { ConnectEnvelopStream } from "./connect-envelop-stream.js";
 
 /**
  * Options used to configure the Connect transport.
@@ -266,15 +268,23 @@ export function createConnectTransport(
 
       async function createRequestBody(
         input: AsyncIterable<I>
-      ): Promise<Uint8Array> {
-        if (method.kind != MethodKind.ServerStreaming) {
+      ): Promise<Uint8Array | ReadableStream> {
+        if (method.kind == MethodKind.ServerStreaming) {
+          const r = await input[Symbol.asyncIterator]().next();
+          if (r.done == true) {
+            throw "missing request message";
+          }
+          return encodeEnvelope(0, serialize(r.value));
+        } else if (
+          method.kind === MethodKind.ClientStreaming &&
+          supportsRequestStreams()
+        ) {
+          const inputStream = iterableToReadableStream(input);
+
+          return inputStream.pipeThrough(new ConnectEnvelopStream(serialize));
+        } else {
           throw "The fetch API does not support streaming request bodies";
         }
-        const r = await input[Symbol.asyncIterator]().next();
-        if (r.done == true) {
-          throw "missing request message";
-        }
-        return encodeEnvelope(0, serialize(r.value));
       }
       return await runStreamingCall<I, O>({
         interceptors: options.interceptors,
@@ -323,3 +333,29 @@ export function createConnectTransport(
     },
   };
 }
+
+function iterableToReadableStream<T>(
+  iterable: AsyncIterable<T>
+): ReadableStream<T> {
+  const it = iterable[Symbol.asyncIterator]();
+  return new ReadableStream<T>(<UnderlyingSource<T>>{
+    async pull(controller: ReadableStreamDefaultController<T>) {
+      const r = await it.next();
+      if (r.done === true) {
+        controller.close();
+        return;
+      }
+      controller.enqueue(r.value);
+    },
+    async cancel(reason) {
+      if (it.throw) {
+        try {
+          await it.throw(reason);
+        } catch {
+          // iterator.throw on a generator function rethrows unless the
+          // body catches and swallows.
+        }
+      }
+    },
+  });
+}

packages/connect-web/src/supports-request-streams.ts

@@ -0,0 +1,15 @@
+// Feature detection function taken from https://developer.chrome.com/articles/fetch-streaming-requests/
+export function supportsRequestStreams() {
+  let duplexAccessed = false;
+
+  const hasContentType = new Request("", {
+    body: new ReadableStream(),
+    method: "POST",
+    get duplex() {
+      duplexAccessed = true;
+      return "half";
+    },
+  } as any).headers.has("Content-Type");
+
+  return duplexAccessed && !hasContentType;
+}