# AsyncAPI

Source: https://apigenie.pl/docs/reference/server/asyncapi
Part of the reference of [apigenie-server](https://apigenie.pl/docs/reference/server.md).
Its other pages: [Responses](https://apigenie.pl/docs/reference/server/responses.md), [Scenarios and extensions](https://apigenie.pl/docs/reference/server/scenarios.md), [Scripts](https://apigenie.pl/docs/reference/server/scripts.md), [Document store](https://apigenie.pl/docs/reference/server/store.md), [Several services](https://apigenie.pl/docs/reference/server/services.md), [Validation and security](https://apigenie.pl/docs/reference/server/validation.md), [Contract testing (validating proxy)](https://apigenie.pl/docs/reference/server/contract-testing.md), [Session recording and replay](https://apigenie.pl/docs/reference/server/recording.md), [Protobuf / gRPC](https://apigenie.pl/docs/reference/server/grpc.md), [GraphQL](https://apigenie.pl/docs/reference/server/graphql.md), [Configuration](https://apigenie.pl/docs/reference/server/configuration.md), [Dashboard and admin API](https://apigenie.pl/docs/reference/server/dashboard.md), [TLS](https://apigenie.pl/docs/reference/server/tls.md).

For an AsyncAPI 3.x document the mock plays the application the document describes. Servers
with the `ws`, `http` and `mqtt` protocols are served (and `wss`, `https` and `mqtts` over TLS
with `--tls-cert`, cleartext without). Kafka and AMQP channels aren't served, but are read from a
real cluster or broker with `--proxy-to kafka://` ([below](#contract-testing-of-kafka-channels))
and `--proxy-to amqp://` ([below](#contract-testing-of-amqp-channels)); the others are listed as not
served.

- **WebSocket.** A client connects to a channel's address under its server's `pathname`
  (`ws://127.0.0.1:4010/v1/rooms/42` for `rooms/{roomId}`). It gets the messages of the
  channel's `send` operations, one after another, every second by default: the message's
  examples in turn, or values generated from its payload schema (always with `--dynamic`). What
  it sends is checked against the `receive` operations: a match is answered with the operation's
  `reply`, anything else with a problem details message.
- **HTTP.** A `receive` operation takes messages with `POST` (or its HTTP binding's method) and
  answers 202, or 200 with its reply; an invalid message gets 422. A `send` operation answers
  `GET` with a message.
- **MQTT.** A broker on its own port takes subscriptions to the channels' topics (wildcards
  included: a parameter of an address takes its default, first enum value or example) and
  publishes the `send` operations' messages with their MQTT binding's QoS. Messages clients
  publish go to the other subscribers, as with any broker, once they match a `receive`
  operation; its reply is published to the reply channel's topic. MQTT 3.1.1 and 5 clients are
  served; MQTT 5 subscribers also get the MQTT 5 properties of a message: its headers as user
  properties (the first example's, else generated from the header schemas), and the payload format
  indicator, content type, correlation data and response topic of its MQTT binding (generated
  from their schemas when they are schemas) and the operation binding's message expiry. An MQTT 5
  client's properties and user properties are checked as the
  [MQTT proxy](#contract-testing-of-mqtt-channels) checks them; its request is answered on the
  topic of its reply address user property, or on its Response Topic when the reply channel has
  `address: null`, with its correlation id and Correlation Data. Retained messages expire with
  their Message Expiry Interval.

`x-apigenie-mock` on a `send` operation sets `interval` (milliseconds), `count` (messages per
client, 0 for no limit), `jitter`, `messages` (the names to send) and `publish: false`.

```yaml
operations:
  sendNotices:
    action: send
    channel: {$ref: '#/channels/room'}
    x-apigenie-mock: {interval: 500, count: 10}
```

The log has `ws-open`, `ws-in`, `ws-out` and `ws-close` events for WebSocket clients and
`mqtt-subscribe`, `mqtt-in`, `mqtt-out` and `mqtt-unsubscribe` ones for MQTT, with the channel's
address, the operation and the message. When the server stops, WebSocket clients get the close
code 1001 and MQTT 5 clients a DISCONNECT; after a reload they reconnect to the new version.

## Contract testing of a WebSocket API

With `--proxy-to ws://host:port` an AsyncAPI document's WebSocket channels aren't mocked: each
client of a channel gets a connection to the real server, and every message of either side is
checked against the document. Point a frontend or a test suite at the mock instead of the server.

```sh
apigenie-server chat.yaml --proxy-to ws://localhost:9000 --report contract.json --fail-on errors
```

The mock connects to the server first, with the client's path under the URL's prefix, its query,
`Origin`, cookies, `Authorization` and `Sec-WebSocket-Protocol`, plus `Via` and `X-Forwarded-*`;
the client gets the subprotocol the server chose. A server that can't be reached answers the
client's handshake with 502 (504 when it doesn't answer in time), and a server that refuses it
(401, 404, ...) has its answer passed on. What clients send is checked against the channel's
`receive` operations, what the server sends against its `send` operations. A client's message of
an operation with a `reply` on the same channel waits for its reply: the first message the server
sends that is one of the reply's messages is checked as that reply (logged as `name (reply)` with
the request's operation), not against the `send` operations. Messages are relayed
whole (fragments are put together first); pings are answered on each side; a close frame goes on
with its code. No WebSocket extension is negotiated (`permessage-deflate` is left out of the
handshake). HTTP channels go to the same server (see [below](#contract-testing-of-http-channels));
the MQTT channels to the broker of a second `--proxy-to mqtt://...`, or they keep being mocked.

| Check | Severity | When |
| --- | --- | --- |
| `message.incompatible` | error | The message matches none of the messages of the channel's operations in its direction. |
| `message.malformed` | error | The message isn't JSON where JSON is expected. |
| `message.frame.unexpected` | warn | A binary frame for a text or JSON message, or a text frame for a binary one (`image/png`, `application/octet-stream`, Protobuf, Avro, ...). |
| `message.qos.unexpected` | warn | MQTT: a client published at a QoS other than the one of the operation's MQTT binding. |
| `channel.parameter.invalid` | error | A parameter of the channel's address (a path segment, a topic level) has a value outside the parameter's `enum`. |
| `channel.receive.undeclared` | warn | A client sent a message on a channel the application has no `receive` operation for. |
| `channel.send.undeclared` | warn | The server sent a message on a channel it has no `send` operation for. |
| `message.reply.uncorrelated` | error | The reply's `correlationId` (in the payload) isn't the request's, or is missing. |
| `operation.query.invalid` | error | HTTP channels: the request's query doesn't match the `query` of the operation's HTTP binding. |
| `message.headers.invalid` | error | HTTP channels: the request's or the reply's header fields don't match the message's `headers` or its HTTP binding's `headers`. |

`validation.request` sets what happens to the clients' messages and `validation.response` to the
server's: `log` by default for a proxy (reported, and forwarded), `enforce` with `--strict`. In
enforce mode a message with an error isn't forwarded, and the client gets a problem message
instead (`asyncapi.validate_incoming` stays the mock's setting):

```json
{"type": "https://apigenie.pl/mock/errors#invalid-upstream-message", "title": "The upstream's message doesn't match the specification",
 "status": 422, "channel": "rooms/1", "direction": "upstream", "validation": [{"location": "message/text", "code": "type",
 "message": "expected type string, found integer", "check": "message.incompatible", "severity": "error"}]}
```

A client's message gets `#invalid-message` with `"direction": "client"`. The log has the
`ws-open`, `ws-in`, `ws-out` and `ws-close` events with the operation, the message's name, the
violations and `source: proxy` (`mock` for what the proxy answered itself). `contract.severity`
changes the checks' severities as for the other proxies.

**Coverage and reports.** Each message of each operation is a branch: it is covered once a valid
one went by. An operation is `covered`, `partial` or `not-tested`; a message that matches none is
counted for the operation it was checked against first. `GET /__apigenie/contract` returns the
report (`"specification": "asyncapi"`, connections, messages in and out, channels, operations,
violations), `--report` and `--report-junit` write it when the server stops (a test case per
operation), and `--fail-on` and `--min-coverage` (the share of messages) make the exit status 5.

## Contract testing of MQTT channels

With `--proxy-to mqtt://host:port` the mock's broker relays to a real broker: each client that
connects to the mock gets a connection of its own to the broker, and every message published
either way is checked against the document.

```sh
apigenie-server devices.yaml --proxy-to mqtt://localhost:1884 --mqtt-port 1883 --report contract.json
apigenie-server events.yaml --proxy-to ws://localhost:9000 --proxy-to mqtt://localhost:1884
```

The upstream connection has the client's MQTT version, client identifier (the one the mock
assigns when the client gives none), clean start, keep-alive, will, user name and password. The
client's CONNECT gets the broker's CONNACK, a refusal included; a broker that can't be reached is
answered with "server unavailable" (return code 3, MQTT 5 0x88). Subscriptions are the broker's:
the SUBACK is its answer, and the broker decides who gets which message. What clients publish is
checked against the topic's channel's `receive` operations, with the QoS of the operation's MQTT
binding when it gives one (`message.qos.unexpected`); what the broker delivers against the `send`
operations. A subscriber also gets other clients' messages, so on a channel without `send`
operations a delivery is checked against the `receive` operations. Topics of no channel are
relayed unchecked (debug level in the log).

In enforce mode a client's message with an error isn't forwarded (MQTT 5 clients get reason code
0x99 in their acknowledgement), and a broker's message with an error isn't delivered. A client
that disconnects is disconnected from the broker; one whose connection is lost has its upstream
connection dropped too, so the broker publishes its will; a client the broker disconnects is
disconnected. A reload keeps the clients connected. The log has `mqtt-connect`,
`mqtt-subscribe`, `mqtt-in`, `mqtt-out`, `mqtt-unsubscribe` and `mqtt-disconnect` events with
`source: proxy` (`mock` for what the proxy decided itself), and coverage and reports work as for
WebSocket.

MQTT 5 properties are relayed both ways: user properties, payload format indicator, content type,
response topic, correlation data and message expiry (not the topic alias and subscription
identifiers, which belong to one connection). An MQTT 5 message's are checked against the MQTT
bindings (0.2.0):

| Binding | Checked (`message.properties.invalid`, `error`) |
| --- | --- |
| message `payloadFormatIndicator` | The Payload Format Indicator is the same (absent counts as 0) |
| message `contentType` | The Content Type is the same, ignoring case |
| message `correlationData` | Correlation Data is present and matches the schema |
| message `responseTopic` | A Response Topic is present, the string given or matching the schema |
| operation `messageExpiryInterval` | A Message Expiry Interval is present and no longer |

Its user properties are its headers, checked against the message's `headers`
(`message.headers.invalid`). MQTT 3.1.1 messages have no properties: nothing is checked for them.
The report counts the messages whose bindings state properties, checked or not:
`summary.properties: {"checked": 3, "unchecked": 1}`, `properties` of each operation, and a
summary line `MQTT 5 properties: checked on 3 messages, not on 1 (MQTT 3.1.1)`. A client's QoS 1
and 2 messages are acknowledged by the mock before the broker
acknowledges them, and the mock delivers at most QoS 1. An `mqtts://` broker (port 8883 by default) is reached over TLS
and verified as `https://` upstreams are (see [TLS](/docs/reference/server/tls)).

## Contract testing of Kafka channels

There is no Kafka broker in the mock. With `--proxy-to kafka://host:port` (port 9092 by
default; `kafkas://`, port 9093, over TLS) the server is a client of the real cluster, a validating
tap: it reads the topics of the document's Kafka channels and checks every record on them,
whoever produced it.

```sh
apigenie-server shop.yaml --proxy-to kafka://localhost:9092 --report contract.json
apigenie-server events.yaml --proxy-to mqtt://localhost:1883 --proxy-to kafka://localhost:9092
```

A channel with a `kafka` or `kafka-secure` server is a Kafka channel; its topic is the channel's
Kafka binding `topic`, else its address. Parameters of the address match the topic's segments, so
`stock.{region}` reads `stock.eu` and `stock.us`; the cluster's topics are looked at again every
10 seconds for new ones. The tap reads every partition from the offset it had when the server
started, outside any consumer group, so it never takes partitions from the application's consumers
and commits no offsets. A topic the cluster doesn't have yet is reported once and read once it
appears; a lost connection is reported and tried again.

A record doesn't say who produced it, so the channel's operations tell. On a channel with only
`send` operations it is the application's (a `kafka-out` event, checked against them); with only
`receive` operations it is a client's (`kafka-in`); with both it is checked as the application's,
and as a client's when that doesn't match. What is checked:

| What | Checked against | Check |
| --- | --- | --- |
| The value | The operations' messages | `message.incompatible`, `message.malformed` |
| The headers | The message's `headers` (values read as the type the schema says) | `message.headers.invalid` |
| The key | The message's Kafka binding `key` (an error too when the record has none) | `message.key.invalid` (`error`) |
| The partition | The channel's Kafka binding `partitions` | `message.partition.invalid` (`warn`) |
| The topic | The address's parameters | `channel.parameter.invalid` |

Records are on the cluster before the tap sees them, so enforce mode reports them like log mode.
Kafka operations count in the coverage, `GET /__apigenie/contract`, `--report` and
`--report-junit` like any other, and the report's `upstream` lists the `kafka://` URL. The log has
`kafka-in` and `kafka-out` events with the topic as `channel`, the record's partition and offset
in the message, and its key (as `kafka-key`) and headers as the event's headers; what the tap
found or failed at are `server` events with `source: kafka`.

The SASL PLAIN user and password come from `APIGENIE_KAFKA_USERNAME` and
`APIGENIE_KAFKA_PASSWORD` (`apigenie test` reads them too), never from the URL. `kafkas://` brokers
are verified as `https://` upstreams are (see [TLS](/docs/reference/server/tls)). The cluster needs Kafka 2.3 or later.
Compressed record batches aren't read: they are reported (`3 record(s) at offset 40 compressed
with zstd`) and skipped. A `kafka://` upstream goes with one document, not with
[several services](/docs/reference/server/services). See ADR-170 for what the client does and doesn't do.

## Contract testing of AMQP channels

There is no AMQP broker in the mock either. With `--proxy-to amqp://host:port/vhost` (port 5672 by
default; `amqps://`, port 5671, over TLS; the path is the virtual host, `/%2Fcarpet-express` for
`/carpet-express`, none for `/`) the server is an AMQP 0-9-1 client of the real broker, a
validating tap.

```sh
APIGENIE_AMQP_USERNAME=conformance APIGENIE_AMQP_PASSWORD=conformance \
  apigenie-server customs.yaml --proxy-to amqp://localhost:5672/%2Fcarpet-express --report contract.json
```

A channel with an `amqp` or `amqps` server is an AMQP channel. Its AMQP channel binding says what it
is:

- **`is: routingKey`** (the default): the address is a routing key of `exchange.name` (of
  `exchange.type`, topic by default). The tap binds a queue of its own (exclusive, deleted when it
  disconnects) to the exchange: for a topic exchange with the address's parameters as `*` words
  (`customs.{country}.filing` is `customs.*.filing`), for a fanout or headers exchange with any key.
  Direct exchanges bind the address itself, so their addresses can't have parameters. It reads that
  queue, so it gets a copy of every message the exchange routes there, and the application's
  queues are left alone. An exchange the broker doesn't have is reported, and looked for again every
  10 seconds.
- **`is: queue`**, and routing keys of the default exchange: a queue the application consumes.
  Reading it would take the application's messages, so the tap doesn't, and says so when it starts.
  What `apigenie test` and the admin API send there is checked.

Messages are checked as the [Kafka tap](#contract-testing-of-kafka-channels) checks records, in the
same direction: their body and headers against the operations' messages, and their properties
against the AMQP bindings (0.3.0) (`message.properties.invalid`, `error`):

| Binding | Checked |
| --- | --- |
| operation `deliveryMode`, `priority`, `userId`, `replyTo` | The property is present and the same |
| operation `expiration` | An expiration is present and no longer |
| operation `timestamp: true` | A timestamp is present |
| message `contentEncoding` | The content encoding is the same |
| message `messageType` | The `type` property is the same |

The log has `amqp-in` and `amqp-out` events with the routing key as `channel` and the exchange in the
message; coverage and reports work as for Kafka, and enforce mode reports too. The PLAIN user and
password come from `APIGENIE_AMQP_USERNAME` and `APIGENIE_AMQP_PASSWORD` (`guest` without them),
never from the URL. An `amqp://` upstream goes with one document. See ADR-171.

## Contract testing of HTTP channels

With `--proxy-to http://host:port` (or the `ws://` server of the WebSocket channels) an AsyncAPI
document's HTTP channels go to that server through the same transport as an OpenAPI document's
proxy (`Via`, `X-Forwarded-*`, 502 and 504 problems), checked against the document instead of
an OpenAPI one:

- A `receive` operation takes requests with its HTTP binding's method (POST by default). The
  request's body is checked against the operation's messages, its query against the binding's
  `query` schema (`operation.query.invalid`) and its header fields against the message's
  `headers` and HTTP binding `headers`, its traits' included (`message.headers.invalid`). A 2XX
  response's body and header fields are checked against its `reply`.
- A `send` operation is asked with GET, as the mock serves it; a 2XX response with a body is
  checked against its messages (not its header fields: they are the webhook's).
- Other paths get 404 and other methods 405, without the server.

Query parameters and header fields are read as the type their schema says (`integer`, `number`,
`boolean`, an `array` from repeated parameters); header fields the schemas don't name are left
out. `apigenie test` checks the requests of HTTP channels with the same code and the same checks
(see [Requests of HTTP channels](/docs/reference/cli/test/validation#requests-of-http-channels)).

In enforce mode an invalid request gets 422 (`#invalid-message`) without the server, and an
invalid response is replaced by 502 (`#invalid-upstream-message`). Responses carry
`X-Apigenie-Mock-Source`, `X-Apigenie-Mock-Operation` and `X-Apigenie-Mock-Violations`, and each
exchange counts as a message of its operation in the coverage. Responses other than 2XX and the
reply's `statusCode` binding aren't checked; fallback mode, `--record-dir` and `--record-har` are
for OpenAPI documents.