server kafkajs

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

Usage

apigenie-generator server 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.

The application is the one the document describes: it sends the messages of send operations and receives those of receive operations. Set the perspective option to counterpart to generate the other side.

import { KafkaApp } from "order-service-kafka-app";

const app = new KafkaApp({ brokers: ["localhost:9092"] }, {
  handleOrderCommands: {
    onPlaceOrderMessage: (message) => ({ payload: { requestId: message.payload.requestId, orderId: "o-1" } }),
    onCancelOrderMessage: () => undefined,
  },
});
await app.start();
await app.publishStockLevels.sendStockLevelMessage({ key: "ABC-0001", payload: { sku: "ABC-0001", quantity: 3 } });
await app.stop();

KafkaApp takes the handlers of all received operations. start() connects the producer and the consumers and resolves once every consumer joined its group; stop() commits, leaves the groups and disconnects. Each sent operation has a sender property.

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 KafkaApp), 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 client 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 server 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.server_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-app. 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:
    server_kafkajs:
      layout: package
      import_extension: .js
      perspective: application
      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	application (default): the described application; counterpart: its counterpart
validation	true (default): validate payloads against their schemas; false: check the headers only

Unknown options are errors.