Add a ConnectRouter based Transport

packages/connect/src/index.ts

@@ -60,3 +60,4 @@ export {
   createMethodImplSpec,
 } from "./implementation.js";
 export type { ServiceImplSpec, MethodImplSpec } from "./implementation.js";
+export { createTransport } from "./protocol-connect/transport.js";

packages/connect/src/protocol/index.ts

@@ -82,6 +82,7 @@ export type {
   UniversalServerRequest,
   UniversalServerResponse,
 } from "./universal.js";
+export { createUniversalHandlerClient } from "./universal-handler-client.js";
 export {
   validateUniversalHandlerOptions,
   createUniversalServiceHandlers,

packages/connect/src/protocol/universal-handler-client.ts

@@ -0,0 +1,65 @@
+// 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 { Code } from "../code.js";
+import { ConnectError } from "../connect-error.js";
+import { createAsyncIterable } from "./async-iterable.js";
+import type { UniversalHandler } from "./universal-handler.js";
+import type { UniversalClientFn } from "./universal.js";
+
+/**
+ * An in-memory UniversalClientFn that can be used to route requests to a ConnectRouter
+ * bypassing network calls. Useful for testing and calling in-process services.
+ */
+export function createUniversalHandlerClient(
+  uHandlers: UniversalHandler[]
+): UniversalClientFn {
+  const handlerMap = new Map<string, UniversalHandler>();
+  for (const handler of uHandlers) {
+    handlerMap.set(handler.requestPath, handler);
+  }
+  return async (uClientReq) => {
+    const reqUrl = new URL(uClientReq.url);
+    const handler = handlerMap.get(reqUrl.pathname);
+    if (!handler) {
+      throw new ConnectError(
+        `RouterHttpClient: no handler registered for ${reqUrl.pathname}`,
+        Code.Unimplemented
+      );
+    }
+    const uServerRes = await handler({
+      body: uClientReq.body,
+      httpVersion: "2.0",
+      method: uClientReq.method,
+      url: reqUrl,
+      header: uClientReq.header,
+    });
+    if (uServerRes.body === undefined) {
+      throw new ConnectError(
+        `RouterHttpClient: missing body in handler response for ${reqUrl.pathname}`,
+        Code.Internal
+      );
+    }
+    let body = uServerRes.body;
+    if (body instanceof Uint8Array) {
+      body = createAsyncIterable([body]);
+    }
+    return {
+      body: body,
+      header: new Headers(uServerRes.header),
+      status: uServerRes.status,
+      trailer: new Headers(uServerRes.trailer),
+    };
+  };
+}

packages/connect/src/routes-transport.spec.ts

@@ -0,0 +1,104 @@
+// 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 { Int32Value, StringValue, MethodKind } from "@bufbuild/protobuf";
+import { createPromiseClient } from "./promise-client.js";
+import { createAsyncIterable } from "./protocol/async-iterable.js";
+import { createRoutesTransport } from "./routes-transport.js";
+
+describe("createRoutesTransport", function () {
+  const testService = {
+    typeName: "TestService",
+    methods: {
+      unary: {
+        name: "Unary",
+        I: Int32Value,
+        O: StringValue,
+        kind: MethodKind.Unary,
+      },
+      server: {
+        name: "Server",
+        I: Int32Value,
+        O: StringValue,
+        kind: MethodKind.ServerStreaming,
+      },
+      client: {
+        name: "Client",
+        I: Int32Value,
+        O: StringValue,
+        kind: MethodKind.ClientStreaming,
+      },
+      biDi: {
+        name: "BiDi",
+        I: Int32Value,
+        O: StringValue,
+        kind: MethodKind.BiDiStreaming,
+      },
+    },
+  } as const;
+  const transport = createRoutesTransport(({ service }) => {
+    service(testService, {
+      unary(req) {
+        return { value: req.value.toString() };
+      },
+      async *server(req) {
+        for (let i = 0; i < req.value; i++) {
+          yield { value: req.value.toString() };
+        }
+      },
+      async client(req) {
+        let value = 0;
+        for await (const next of req) {
+          value = next.value;
+        }
+        return { value: value.toString() };
+      },
+      async *biDi(req) {
+        for await (const next of req) {
+          yield { value: next.value.toString() };
+        }
+      },
+    });
+  });
+  const client = createPromiseClient(testService, transport);
+  it("should work for unary", async function () {
+    const res = await client.unary({ value: 13 });
+    expect(res.value).toBe("13");
+  });
+  it("should work for server steam", async function* () {
+    const res = client.server({ value: 13 });
+    let count = 0;
+    for await (const next of res) {
+      count++;
+      expect(next.value).toBe("13");
+    }
+    expect(count).toBe(13);
+  });
+  it("should work for client steam", async function () {
+    const res = await client.client(
+      createAsyncIterable([{ value: 12 }, { value: 13 }])
+    );
+    expect(res.value).toBe("13");
+  });
+  it("should work for bidi steam", async function () {
+    const payload = [{ value: 1 }, { value: 2 }];
+    const res = client.biDi(createAsyncIterable(payload));
+    let count = 0;
+    for await (const next of res) {
+      expect(next.value).toBe(payload[count].value.toString());
+      count++;
+    }
+    expect(count).toBe(payload.length);
+  });
+});

packages/connect/src/routes-transport.ts

@@ -0,0 +1,55 @@
+// 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 { createTransport } from "./protocol-connect/transport.js";
+import type { CommonTransportOptions } from "./protocol/transport-options.js";
+import { createUniversalHandlerClient } from "./protocol/universal-handler-client.js";
+import {
+  ConnectRouter,
+  ConnectRouterOptions,
+  createConnectRouter,
+} from "./router.js";
+
+/**
+ * Creates a Transport that routes requests to the configured routes. Useful for testing
+ * and calling services running in the same process.
+ *
+ * This can be used to test both client logic by using this to stub/mock the backend,
+ * and to test server logic by using this to run without needing to spin up a server.
+ */
+export function createRoutesTransport(
+  routes: (router: ConnectRouter) => void,
+  options?: {
+    transport?: Partial<CommonTransportOptions>;
+    router?: ConnectRouterOptions;
+  }
+) {
+  const router = createConnectRouter({
+    ...(options?.router ?? {}),
+    connect: true,
+  });
+  routes(router);
+  return createTransport({
+    httpClient: createUniversalHandlerClient(router.handlers),
+    baseUrl: "https://in-memory",
+    useBinaryFormat: true,
+    interceptors: [],
+    acceptCompression: [],
+    sendCompression: null,
+    compressMinBytes: Number.MAX_SAFE_INTEGER,
+    readMaxBytes: Number.MAX_SAFE_INTEGER,
+    writeMaxBytes: Number.MAX_SAFE_INTEGER,
+    ...(options?.transport ?? {}),
+  });
+}