# server kafkajs

Source: https://apigenie.pl/techdocs/reference/cli-generate_server_kafkajs

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

## Usage

```sh
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](https://kafka.js.org) 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.

```ts
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`](/techdocs/reference/cli-generate_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`:

```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
```

```sh
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

```yaml
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.