Add support for server side interceptors

packages/connect-web-bench/README.md

@@ -10,5 +10,5 @@ it like a web server would usually do.
 
 | code generator | bundle size        | minified               | compressed           |
 |----------------|-------------------:|-----------------------:|---------------------:|
-| connect        | 117,695 b | 51,217 b | 13,730 b |
+| connect        | 117,734 b | 51,217 b | 13,696 b |
 | grpc-web       | 415,212 b    | 300,936 b    | 53,420 b |

packages/connect/src/implementation.spec.ts

@@ -12,8 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-import { createHandlerContext } from "./implementation.js";
+import {
+  createHandlerContext,
+  createMethodImplSpec,
+} from "./implementation.js";
+import type { MethodImplSpec } from "./implementation.js";
 import { Int32Value, MethodKind, StringValue } from "@bufbuild/protobuf";
+import { createAsyncIterable } from "./protocol/async-iterable.js";
+import { createContextKey } from "./context-values.js";
 
 const TestService = {
   typeName: "handwritten.TestService",
@@ -24,6 +30,24 @@ const TestService = {
       O: StringValue,
       kind: MethodKind.Unary,
     },
+    clientStreaming: {
+      name: "ClientStreaming",
+      I: Int32Value,
+      O: StringValue,
+      kind: MethodKind.ClientStreaming,
+    },
+    serverStreaming: {
+      name: "ServerStreaming",
+      I: Int32Value,
+      O: StringValue,
+      kind: MethodKind.ServerStreaming,
+    },
+    bidiStreaming: {
+      name: "BidiStreaming",
+      I: Int32Value,
+      O: StringValue,
+      kind: MethodKind.BiDiStreaming,
+    },
   },
 } as const;
 
@@ -33,6 +57,7 @@ describe("createHandlerContext()", function () {
     method: TestService.methods.unary,
     protocolName: "foo",
     requestMethod: "GET",
+    url: "https://example.com/foo",
   };
 
   describe("signal", function () {
@@ -112,3 +137,223 @@ describe("createHandlerContext()", function () {
     expect(ctx.responseTrailer.get("foo")).toBe("trailer");
   });
 });
+
+describe("createMethodImplSpec()", () => {
+  const kFoo = createContextKey("foo");
+  const handlerInit = {
+    service: TestService,
+    url: "https://example.com/foo",
+    protocolName: "connect",
+    requestMethod: "POST",
+    requestHeader: {
+      Key: "Value",
+    },
+  };
+  it("should apply interceptors to unary calls", async () => {
+    const spec = createMethodImplSpec(
+      TestService,
+      TestService.methods.unary,
+      (_, { responseHeader, responseTrailer }) => {
+        responseHeader.set("Key", "bar");
+        responseTrailer.set("TKey", "tbar");
+        return { value: "foo" };
+      },
+      [
+        (next) => async (req) => {
+          expect(req.stream).toEqual(false);
+          expect(req.init.method).toEqual("POST");
+          expect(req.service).toEqual(TestService);
+          expect(req.header.get("Key")).toEqual("Value");
+          expect(req.url).toEqual("https://example.com/foo");
+          expect(req.method).toEqual(TestService.methods.unary);
+          expect(context.values.get(kFoo)).toEqual("bar");
+          req.header.set("Key", "bar");
+          const res = await next(req);
+          expect(res.stream).toEqual(false);
+          expect(res.service).toEqual(TestService);
+          expect(res.header.get("Key")).toEqual("bar");
+          expect(res.method).toEqual(TestService.methods.unary);
+          res.header.set("Key", "baz");
+          return res;
+        },
+        (next) => async (req) => {
+          expect(req.header.get("Key")).toEqual("bar");
+          return next(req);
+        },
+      ],
+    ) as MethodImplSpec<Int32Value, StringValue> & {
+      kind: MethodKind.Unary;
+    };
+    const context = createHandlerContext({
+      ...handlerInit,
+      method: TestService.methods.unary,
+    });
+    context.values.set(kFoo, "bar");
+    const res = await spec.impl(new Int32Value({ value: 1 }), context);
+    expect(res).toEqual(new StringValue({ value: "foo" }));
+    expect(context.responseHeader.get("Key")).toEqual("baz");
+    expect(context.responseTrailer.get("TKey")).toEqual("tbar");
+  });
+  it("should apply interceptors to client streaming calls", async () => {
+    const spec = createMethodImplSpec(
+      TestService,
+      TestService.methods.clientStreaming,
+      // eslint-disable-next-line @typescript-eslint/require-await
+      async (_, { responseHeader, responseTrailer }) => {
+        responseHeader.set("Key", "bar");
+        responseTrailer.set("TKey", "tbar");
+        return { value: "foo" };
+      },
+      [
+        (next) => async (req) => {
+          expect(req.stream).toEqual(true);
+          expect(req.init.method).toEqual("POST");
+          expect(req.service).toEqual(TestService);
+          expect(req.header.get("Key")).toEqual("Value");
+          expect(req.url).toEqual("https://example.com/foo");
+          expect(req.method).toEqual(TestService.methods.clientStreaming);
+          expect(context.values.get(kFoo)).toEqual("bar");
+          req.header.set("Key", "bar");
+          const res = await next(req);
+          expect(res.stream).toEqual(false);
+          expect(res.service).toEqual(TestService);
+          expect(res.header.get("Key")).toEqual("bar");
+          expect(res.method).toEqual(TestService.methods.clientStreaming);
+          res.header.set("Key", "baz");
+          return res;
+        },
+        (next) => async (req) => {
+          expect(req.header.get("Key")).toEqual("bar");
+          return next(req);
+        },
+      ],
+    ) as MethodImplSpec<Int32Value, StringValue> & {
+      kind: MethodKind.ClientStreaming;
+    };
+    const context = createHandlerContext({
+      ...handlerInit,
+      method: TestService.methods.clientStreaming,
+    });
+    context.values.set(kFoo, "bar");
+    const res = await spec.impl(
+      createAsyncIterable([new Int32Value({ value: 1 })]),
+      context,
+    );
+    expect(res).toEqual(new StringValue({ value: "foo" }));
+    expect(context.responseHeader.get("Key")).toEqual("baz");
+    expect(context.responseTrailer.get("TKey")).toEqual("tbar");
+  });
+  it("should apply interceptors to server streaming calls", async () => {
+    const spec = createMethodImplSpec(
+      TestService,
+      TestService.methods.serverStreaming,
+      // eslint-disable-next-line @typescript-eslint/require-await
+      async function* (_, { responseHeader, responseTrailer }) {
+        responseHeader.set("Key", "bar");
+        responseTrailer.set("TKey", "tbar");
+        yield { value: "foo" };
+      },
+      [
+        (next) => async (req) => {
+          expect(req.stream).toEqual(false);
+          expect(req.init.method).toEqual("POST");
+          expect(req.service).toEqual(TestService);
+          expect(req.header.get("Key")).toEqual("Value");
+          expect(req.url).toEqual("https://example.com/foo");
+          expect(req.method).toEqual(TestService.methods.serverStreaming);
+          expect(context.values.get(kFoo)).toEqual("bar");
+          req.header.set("Key", "bar");
+          const res = await next(req);
+          const responses = [];
+          for await (const next of res.message as AsyncIterable<StringValue>) {
+            responses.push(next);
+          }
+          expect(res.stream).toEqual(true);
+          expect(res.service).toEqual(TestService);
+          expect(res.header.get("Key")).toEqual("bar");
+          expect(res.method).toEqual(TestService.methods.serverStreaming);
+          res.header.set("Key", "baz");
+          return {
+            ...res,
+            message: createAsyncIterable(responses),
+          } as typeof res;
+        },
+        (next) => async (req) => {
+          expect(req.header.get("Key")).toEqual("bar");
+          return next(req);
+        },
+      ],
+    ) as MethodImplSpec<Int32Value, StringValue> & {
+      kind: MethodKind.ServerStreaming;
+    };
+    const context = createHandlerContext({
+      ...handlerInit,
+      method: TestService.methods.serverStreaming,
+    });
+    context.values.set(kFoo, "bar");
+    const res = spec.impl(new Int32Value({ value: 1 }), context);
+    for await (const next of res) {
+      expect(next).toEqual(new StringValue({ value: "foo" }));
+    }
+    expect(context.responseHeader.get("Key")).toEqual("baz");
+    expect(context.responseTrailer.get("TKey")).toEqual("tbar");
+  });
+  it("should apply interceptors to bidi streaming calls", async () => {
+    const spec = createMethodImplSpec(
+      TestService,
+      TestService.methods.bidiStreaming,
+      // eslint-disable-next-line @typescript-eslint/require-await
+      async function* (_, { responseHeader, responseTrailer }) {
+        responseHeader.set("Key", "bar");
+        responseTrailer.set("TKey", "tbar");
+        yield { value: "foo" };
+      },
+      [
+        (next) => async (req) => {
+          expect(req.stream).toEqual(true);
+          expect(req.init.method).toEqual("POST");
+          expect(req.service).toEqual(TestService);
+          expect(req.header.get("Key")).toEqual("Value");
+          expect(req.url).toEqual("https://example.com/foo");
+          expect(req.method).toEqual(TestService.methods.bidiStreaming);
+          expect(context.values.get(kFoo)).toEqual("bar");
+          req.header.set("Key", "bar");
+          const res = await next(req);
+          const responses = [];
+          for await (const next of res.message as AsyncIterable<StringValue>) {
+            responses.push(next);
+          }
+          expect(res.stream).toEqual(true);
+          expect(res.service).toEqual(TestService);
+          expect(res.header.get("Key")).toEqual("bar");
+          expect(res.method).toEqual(TestService.methods.bidiStreaming);
+          res.header.set("Key", "baz");
+          return {
+            ...res,
+            message: createAsyncIterable(responses),
+          } as typeof res;
+        },
+        (next) => async (req) => {
+          expect(req.header.get("Key")).toEqual("bar");
+          return next(req);
+        },
+      ],
+    ) as MethodImplSpec<Int32Value, StringValue> & {
+      kind: MethodKind.BiDiStreaming;
+    };
+    const context = createHandlerContext({
+      ...handlerInit,
+      method: TestService.methods.bidiStreaming,
+    });
+    context.values.set(kFoo, "bar");
+    const res = spec.impl(
+      createAsyncIterable([new Int32Value({ value: 1 })]),
+      context,
+    );
+    for await (const next of res) {
+      expect(next).toEqual(new StringValue({ value: "foo" }));
+    }
+    expect(context.responseHeader.get("Key")).toEqual("baz");
+    expect(context.responseTrailer.get("TKey")).toEqual("tbar");
+  });
+});