Transport Adapters

Adapters provide protocol-specific implementations of the Transport interface.

Available Adapters#

---

Transport Interface#

All adapters implement:

interface Transport {
  readonly name: string;
  send<TReq, TRes>(message: Message<TReq>): AsyncIterable<ResponseItem<TRes>>;
  close(): Promise<void>;
}

---

HTTP Adapter#

Client#

import { HttpTransport } from "client/adapters/http/client";

const transport = new HttpTransport({
  baseUrl: "https://api.example.com",

  // URL strategy (how to map method → URL)
  urlStrategy: defaultUrlPattern,  // or custom

  // HTTP method strategy (how to map operation → HTTP verb)
  httpMethodStrategy: restfulHttpMethodStrategy,  // or postOnlyStrategy

  // Default headers
  headers: {
    "Content-Type": "application/json",
  },

  // Request timeout (ms)
  timeout: 30000,

  // Custom fetch implementation
  fetch: customFetch,
});

const client = new Client(transport);

URL Strategies#

// Default: /{version?}/{service}/{operation}
defaultUrlPattern(method)
// { service: "users", operation: "get" } → "/users/get"
// { service: "users", operation: "get", version: "v2" } → "/v2/users/get"

// RESTful: /{version?}/{service}
restfulUrlPattern(method)
// { service: "users", operation: "get" } → "/users"

// Custom
const customUrlStrategy = (method: Method) => {
  return `/api/${method.service}/${method.operation}`;
};

HTTP Method Strategies#

// RESTful mapping (default)
restfulHttpMethodStrategy(method)
// "get" → GET
// "list" → GET
// "create" → POST
// "update" → PUT
// "patch" → PATCH
// "delete" → DELETE
// other → POST

// POST only
postOnlyStrategy(method)
// All operations → POST

Server#

import express from "express";
import { HttpServerTransport } from "@procedures/client";

const app = express();
// A body parser is optional: without one, the transport reads the JSON body itself.
// With one, use strict: false, so that a number, a string or null can be a payload.
app.use(express.json({ strict: false }));

const transport = new HttpServerTransport(server, {
  app,                 // required
  port: 3000,
  host: "127.0.0.1",   // the default: loopback only

  // The default URL strategy parses the HTTP client's format, /api/{service}/{operation}:
  // leave it out unless the client uses another format
  // urlStrategy: rpcServerUrlStrategy,

  // CORS is off by default. origin is "*", one origin or a list.
  cors: true,
  corsOptions: { origin: ["https://app.example"] },
});

server.addTransport(transport);

Server URL Strategies#

// Default (createPatternServerUrlStrategy): the inverse of the HTTP client's URL format
// POST /api/users/get → { service: "users", operation: "get" }
// POST /api/v2/users/create → { service: "users", operation: "create", version: "v2" }
// POST /api/docker.compose/up → { service: "docker.compose", operation: "up" }

// RESTful: the operation comes from the HTTP verb
defaultServerUrlStrategy(req)
// GET /api/users → { service: "users", operation: "list" }

// RPC style
rpcServerUrlStrategy(req)
// POST /rpc/users/get → { service: "users", operation: "get" }

A procedure path maps to a method with pathToMethod: the last segment is the operation.

---

WebSocket Adapter#

Client#

import { WebSocketTransport } from "client/adapters/websocket/client";

const transport = new WebSocketTransport({
  url: "wss://api.example.com/ws",

  // Auto-reconnect settings
  reconnect: true,
  reconnectInterval: 1000,
  maxReconnectAttempts: 5,

  // Connection timeout
  timeout: 30000,

  // Message serialization
  serialize: JSON.stringify,
  deserialize: JSON.parse,

  // Auth token for initial connection
  authToken: "bearer-token",
});

const client = new Client(transport);

Connection Lifecycle#

// Events (if supported by implementation)
transport.on("connect", () => console.log("Connected"));
transport.on("disconnect", () => console.log("Disconnected"));
transport.on("reconnect", (attempt) => console.log(`Reconnecting: attempt ${attempt}`));
transport.on("error", (error) => console.error("Error:", error));

Server#

import { WebSocketServerTransport } from "client/adapters/websocket/server";

const transport = new WebSocketServerTransport({
  port: 3001,

  // Authentication handler
  onAuth: async (token) => {
    const user = await verifyToken(token);
    if (!user) throw new Error("Invalid token");
    return user;  // Attached to connection
  },

  // Connection handlers
  onConnection: (socket, user) => {
    console.log(`User ${user.id} connected`);
  },
  onDisconnection: (socket, user) => {
    console.log(`User ${user.id} disconnected`);
  },

  // Message handlers
  onMessage: (socket, message) => { ... },
  onError: (socket, error) => { ... },
});

server.addTransport(transport);

Streaming#

WebSocket naturally supports streaming:

// Client streaming
for await (const event of client.stream(
  { service: "events", operation: "watch" },
  { topic: "orders" }
)) {
  console.log("Event:", event);
}

---

Local Adapter#

In-process RPC for testing and embedded use.

Client#

import { LocalTransport } from "client/adapters/local/client";

// Create handlers map
const handlers = new Map<string, (payload: unknown) => Promise<unknown>>();

handlers.set("users.get", async (payload: { id: string }) => {
  return { id: payload.id, name: "John Doe" };
});

handlers.set("users.create", async (payload: { name: string }) => {
  return { id: "123", name: payload.name };
});

// Create transport
const transport = new LocalTransport({
  handlers,

  // Optional: simulate latency
  latency: 10,  // ms

  // Optional: simulate errors
  errorRate: 0.01,  // 1% of requests fail
});

const client = new Client(transport);

// Use just like HTTP client
const user = await client.call(
  { service: "users", operation: "get" },
  { id: "123" }
);

Use Cases#

1. Testing: Test client code without network
2. Development: Develop UI before backend is ready
3. Embedded: Same-process RPC for plugin systems
4. Migration: Gradual transition from monolith to microservices

---

Mock Adapter#

For unit testing with predefined responses.

Client#

import { MockTransport, mockBuilder } from "client/adapters/mock/client";

// Build mock responses
const mock = mockBuilder()
  // Exact match
  .when({ service: "users", operation: "get" }, { id: "123" })
  .respond({ id: "123", name: "John Doe" })

  // Pattern match
  .when({ service: "users", operation: /create|update/ })
  .respond((payload) => ({ id: "new", ...payload }))

  // Error response
  .when({ service: "users", operation: "delete" })
  .error("FORBIDDEN", "Not allowed")

  // Delay
  .when({ service: "slow", operation: "endpoint" })
  .delay(1000)
  .respond({ result: "ok" })

  .build();

const transport = new MockTransport(mock);
const client = new Client(transport);

// Use in tests
const user = await client.call(
  { service: "users", operation: "get" },
  { id: "123" }
);
expect(user.name).toBe("John Doe");

Verifying Calls#

const transport = new MockTransport(mock);

// ... make calls ...

// Get call history
const calls = transport.getCalls();
expect(calls).toHaveLength(3);
expect(calls[0].method).toEqual({ service: "users", operation: "get" });
expect(calls[0].payload).toEqual({ id: "123" });

// Verify specific call was made
expect(transport.wasCalled({ service: "users", operation: "get" })).toBe(true);
expect(transport.getCallCount({ service: "users", operation: "get" })).toBe(1);

// Clear history
transport.clearCalls();

Test Example#

describe("UserService", () => {
  let client: Client;
  let transport: MockTransport;

  beforeEach(() => {
    transport = new MockTransport(
      mockBuilder()
        .when({ service: "users", operation: "get" }, { id: "123" })
        .respond({ id: "123", name: "Test User", email: "test@example.com" })
        .build()
    );
    client = new Client(transport);
  });

  afterEach(() => {
    transport.clearCalls();
  });

  it("should fetch user by id", async () => {
    const user = await client.call(
      { service: "users", operation: "get" },
      { id: "123" }
    );

    expect(user).toEqual({
      id: "123",
      name: "Test User",
      email: "test@example.com",
    });
    expect(transport.getCallCount({ service: "users", operation: "get" })).toBe(1);
  });
});

---

Choosing an Adapter#

---

Custom Adapters#

Implement the Transport interface:

class CustomTransport implements Transport {
  readonly name = "custom";

  async *send<TReq, TRes>(message: Message<TReq>): AsyncIterable<ResponseItem<TRes>> {
    // 1. Convert message to protocol format
    const wireFormat = this.serialize(message);

    // 2. Send over your protocol
    const response = await this.protocol.send(wireFormat);

    // 3. Convert response to ResponseItem
    yield {
      id: message.id,
      status: { type: "success", code: 200 },
      payload: response.data as TRes,
      metadata: {},
    };
  }

  async close(): Promise<void> {
    await this.protocol.disconnect();
  }
}