API Genie

Generate a TypeScript (Node.js) Kafka client of an AsyncAPI document, on KafkaJS.

Usage

apigenie-generator client kafkajs [...OPTIONS] INPUT_FILE

Description

The input must be an AsyncAPI 3.x document. Documents referencing other files are bundled first.

The generator handles the Kafka channels: channels of a server whose protocol is kafka or kafka-secure, or with a kafka binding when the document has no servers. Other channels are skipped with an info message. The generated code runs on Node.js with KafkaJS 2.2.4, its only dependency.

An AsyncAPI document describes an application. By default the client is its counterpart: it sends the messages of receive operations and receives the messages of send operations. Set the perspective option to application to generate the application’s side instead.

import { KafkaClient } from "order-service-kafka-client";

const client = new KafkaClient({ brokers: ["localhost:9092"] });
const reply = await client.handleOrderCommands.sendPlaceOrderMessage({
  payload: { type: "place", requestId: crypto.randomUUID(), items: [{ sku: "ABC-0001", quantity: 1 }] },
});
await client.listenPublishStockLevels({
  onStockLevelMessage: (message) => console.log(message.payload.sku),
});
await client.close();

KafkaClient has connect() (the senders connect on first use otherwise), a sender property per sent operation, a listen<Operation>(handler) method per received operation, which resolves once its consumer joined its group, and close().

Every sent operation gets a sender property with a send<Message>(message) method per message. It validates the message, sets the constant headers (e.g. ce_type) and resolves once the brokers have it. A message is { key?, headers, payload }, with the headers under their names on the wire (ce_id). The method of an operation with a reply sends the request with a new correlation id unless it has one, and resolves with the reply: the reply message, or { name, message } when there are several. It rejects with ReplyTimeoutError when no reply arrives in time ({ timeoutMs }, KafkaOptions.replyTimeoutMs, 30 seconds by default).

Every received operation gets a handler interface: on<Message>(message, record) for each of its messages and an optional onUnknown(record). Records are matched to messages by their ce_type (or other constant) header, by a constant payload property, or, failing both, by the first message whose payload is valid. The handlers of an operation with a reply return the reply (or undefined for none), which is sent with the request’s correlation id. Invalid records are reported to KafkaOptions.onError and skipped. A handler that throws is reported too and the record is retried (KafkaJS doesn’t commit it).

Payloads and headers are validated against their schemas by generated functions without dependencies; invalid messages throw MessageValidationError, whose issues point into /headers or /payload. KafkaOptions takes the brokers, the producer’s clientId, config (KafkaJS options for every client: ssl, sasl, retry, logLevel, …), groupIds replacing the generated consumer groups, fromBeginning (true by default) and onError. The generated code creates no topics.

The package contains src/types.ts (payload types), src/messages.ts (message and header interfaces), src/kafka.ts (servers, topics, ids, senders, handlers and the KafkaClient), src/codecs.ts (validators and the descriptions of the messages the runtime works with), src/runtime.ts, src/validation_runtime.ts and src/index.ts, with package.json and tsconfig.json.

Not supported, and reported as warnings: Avro payloads, topics with parameters, replies at runtime-expression addresses, and replies without correlation ids. No security code is generated (pass KafkaJS’s ssl and sasl in KafkaOptions.config).

apigenie-generator server kafkajs generates the other side on the same runtime. The design is ADR-120.

Extensions

Extension	Effect
`x-apigenie.codegen.name`	name of the type of a component schema
`x-apigenie.codegen.disabled`	`true` on a channel or an operation leaves it out (with the operations of a channel left out)

The extensions can live in an overlay for this target instead of in the document, e.g. kafkajs.overlay.yaml:

overlay: 1.0.0
info:
  title: TypeScript names
  version: 1.0.0
actions:
  - target: $.components.schemas.Order
    update:
      x-apigenie:
        codegen:
          name: PurchaseOrder
  - target: $.operations.publishAudit
    update:
      x-apigenie:
        codegen:
          disabled: true

apigenie-generator client kafkajs asyncapi.yaml --overlay kafkajs.overlay.yaml -o out

Options

-c, –config

Path to the configuration file to use. See the “Configuration” section for more information.

–overlay OVERLAY

Path to an OpenAPI Overlay (1.0 to 1.2) applied to the input before generating, e.g. one setting the extensions of the “Extensions” section. Can be repeated: the overlays apply in order, after those listed under commands.generate.client_kafkajs.overlay in the configuration file (with paths relative to it). Warnings and errors about a value an overlay set point to the overlay. The overlays apply to the AsyncAPI document bundled with the files it references.

–strict

Makes an overlay action whose target selects nothing an error (exit status 2, nothing is generated) instead of a warning (OVERLAY:LINE:COL: action #N (TARGET) selects no nodes.).

-o, –output

Directory where the package is generated.

–package-name

Name of the generated package. Defaults to the title of the API followed by -kafka-client. It is also the base of the generated consumer group and client ids.

–package-version

Version of the generated package. Defaults to the version of the API.

Configuration

version: '1.0'
commands:
  generate:
    client_kafkajs:
      layout: package
      import_extension: .js
      perspective: counterpart
      validation: true

Option	Description
`layout`	`package` (default): an npm package with sources in `src/`; `sources`: the sources only
`import_extension`	Extension of relative imports: `.js` (default, Node.js) or an empty string (bundlers)
`perspective`	`counterpart` (default): the counterpart of the described application; `application`: the application itself
`validation`	`true` (default): validate payloads against their schemas; `false`: check the headers only

Unknown options are errors.