# Publishing and consuming on a broker

> Goal: let an HTTP service publish to a Kafka, AMQP 0-9-1, MQTT, NATS or AMQP 1.0 broker without embedding a client, and let a browser, a mobile app or a…

Guide - Since version 1.5.0 - Flexway v1.6.3 - https://flexway.io/en/docs/operations-async/

> Operations guide: configure, deploy and observe the gateway. What the capability does, product side: [product guide](https://flexway.io/en/docs/product-overview/); calling it: [developer guide](https://flexway.io/en/docs/developer-overview/).

**Goal**: let an HTTP service publish to a Kafka, AMQP 0-9-1, MQTT, NATS or AMQP 1.0 broker **without embedding a
client**, and let a browser, a mobile app or a partner receive **its tenant's** events - behind the
same controls as an API: identity, scopes, rate, quota, trace, audit.

Three files, one role each: the **AsyncAPI** document says the channels and messages, the **OpenAPI**
contract says the HTTP doors, `flexway.yaml` says where the brokers are.

```yaml
# flexway.yaml - lists no channel: it references the contract
contracts:
  openapi: ./api/openapi.yaml
  asyncapi: ./api/asyncapi.yaml        # AsyncAPI 3.0.0 or 3.1.0
upstreams:
  events:
    protocol: kafka                    # or amqp (amqps://...), mqtt (mqtts://..., protocol_version: 5 or 3.1.1)
    brokers: [kafka-0.kafka:9093, kafka-1.kafka:9093]
    tls: { ca_file: /etc/flexway/kafka/ca.crt }
    sasl: { mechanism: SCRAM-SHA-512, username: flexway, password: "file:/etc/flexway-secrets/kafka/password" }
    timeout: 2s                        # a publication's budget, acknowledgement included
    max_in_flight: 512                 # bounded buffer: past it, 503 overloaded
  partner-hook:
    protocol: webhook
    url: https://hooks.partner.example/orders
    secrets: ["file:/etc/flexway-secrets/hooks/current", "file:/etc/flexway-secrets/hooks/next"]   # whsec_<base64>
    retries: { max: 8, max_delay: 10m }
```

```yaml
# api/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Orders events, version: 1.0.0 }
channels:
  orderCreated:
    address: "orders.{tenant}.created"
    parameters:
      tenant: { x-flexway-bind: tenant }     # bound to the identity, never read from the request
    messages:
      created: { payload: { $ref: "./schemas/order-created.json" } }
operations:
  publishOrderCreated:
    action: send
    channel: { $ref: "#/channels/orderCreated" }
    x-flexway-upstream: events
    x-flexway-key: /orderId                  # order is kept per key, and per key only
  receiveOrderCreated:
    action: receive
    channel: { $ref: "#/channels/orderCreated" }
    x-flexway-upstream: events
    x-flexway-deliver:
      - { upstream: partner-hook, tenant: acme, dead_letter: orders.dead }
```

```yaml
# api/openapi.yaml - the doors, like any operation
paths:
  /orders/events:
    post:
      operationId: emitOrderCreated
      x-flexway-publish: { operation: publishOrderCreated, idempotency: required }
      x-flexway-tool: { name: emit_order_created }      # optional: an MCP tool too
      requestBody:
        content:                       # inline: the OpenAPI contract does not follow a $ref to a file
          application/json: { schema: { type: object, required: [orderId], properties: { orderId: { type: string } } } }
      responses: { "202": { description: Acknowledged by the broker } }
  /orders/stream:
    get:
      operationId: streamOrderCreated
      x-flexway-subscribe: { operation: receiveOrderCreated, transport: [sse, websocket], max_streams: 2, ack_timeout: 30s, history: 64 }
      responses: { "200": { description: The tenant's events } }
```

**What happens**:

| Step | What the gateway does |
| --- | --- |
| Publish (`POST /orders/events`) | The whole admission, the body checked against the message's schema, a **CloudEvents 1.0** event (`id` = `Idempotency-Key`, `tenant` from the identity, `traceparent`), **`202` after the broker's acknowledgement**, with `{"id": ...}` |
| Subscribe over SSE | `GET` with `Accept: text/event-stream`: one event per message of **its** tenant. A replica reads each channel **once**, whatever its subscribers (`flexway_async_consumers`), and hands each its own; a subscriber too slow for its `buffer` is closed (`overloaded`, `flexway_async_disconnects_total{reason="slow"}`). A client back with `Last-Event-ID` resumes from the channel's history (`history`, 64 messages); beyond it, `412 precondition_failed`: open again without `Last-Event-ID` |
| Subscribe over WebSocket | Each message is a JSON CloudEvent; the client acknowledges `{"ack": "<id>"}`; without an acknowledgement within `ack_timeout`, close `1008` and delivered again on return - **at least once** |
| Deliver to a webhook | One replica per message, signed Standard Webhooks (`webhook-id` = `id`), bounded retries, then the **dead letter** with `problemcode` and `problemstatus` |

**What to know**:

- **`504` = unknown outcome**: the broker did not acknowledge within the budget, the message may have
  been written. Send it again with **the same** `Idempotency-Key`: consumers deduplicate on `source` +
  `id`. `idempotency: required` refuses a publication without one (`428`).
- **Order**: per key only (`x-flexway-key`). A slow webhook holds its partition, never the other
  deliveries - bounded by its breaker and its retries.
- **The tenant**: a channel bound to the tenant (`x-flexway-bind`) or a message declaring the `tenant`
  header - with neither, the consumption is **refused at start**. A `ce-*` header sent by the caller is
  refused (`400`): the attributes are the gateway's.
- **A channel per person**: `x-flexway-bind: subject` (or `claim:<name>`, a claim of the verified
  token) - `players.{sub}.notifications` is read by that subject alone; a value outside letters,
  digits, `_` and `-` is refused (`403`).
- **A public feed** (anonymous viewers, a display screen): `x-flexway-public: true` on the `receive`
  operation, a door with `security: []`, `max_anonymous_streams` to bound every anonymous caller
  together; refused when the channel binds or filters anything.
- **Structured mode**: a producer outside the gateway (an outbox) may write
  `application/cloudevents+json`; the gateway reads both modes, by the content type.
- **Minimal broker ACLs**, for the gateway's principal: write the topics of the `send` operations (and
  the dead letter), read those of the `receive` ones, manage its `flexway.ws.*` and `flexway.deliver.*`
  groups (Kafka) or declare its queues (AMQP). **Never** create a topic: the gateway creates none - a
  tenant without its topic answers `502 upstream_error`.
- **An unreachable broker** never blocks the start: its breaker opens (`503`), other operations
  serve. An unchanged client is kept across reloads.

**Check**:

```bash
curl -i -X POST http://localhost:8080/orders/events -H "Authorization: Bearer $TOKEN" \
  -H 'Content-Type: application/json' -H 'Idempotency-Key: order-42' -d '{"orderId":"42"}'
# HTTP/1.1 202 Accepted  {"id":"order-42"}
curl -N http://localhost:8080/orders/stream -H "Authorization: Bearer $TOKEN" -H 'Accept: text/event-stream'
# id: 0:18,1:7,2:11
# event: created
# data: {"specversion":"1.0","id":"order-42","tenant":"acme","data":{"orderId":"42"},...}
```

`/config` (admin) shows `async`: channels, direction, isolation (`bind`, `public`), doors (publish or
subscribe, transports, caps) and deliveries, never an address resolved for a tenant;
`flexway_async_publish_total` and `flexway_async_deliver_total` count by operation, tenant and
outcome; a publication waiting for its acknowledgement counts in the broker's
`flexway_requests_in_flight`.

**Binary payloads** (`1.6.0`): a message may be **Avro** or **Protobuf** (`payload.schemaFormat`)
or **CBOR** (`contentType: application/cbor` and a JSON schema). The schema is compiled at load; a
publication requires the declared `Content-Type` (`415` otherwise, `422` when the door declares another
body) and decodes the body under its schema when the AsyncAPI operation asks for it
(`x-flexway-validation: full`, the operation's choice: one decoding per message); an SSE subscriber or a
webhook receives the message **in JSON** (`datacontenttype: application/json`), a WebSocket receives it as
it is when its door declares `frames: binary`. Confluent's wire format (magic byte, schema id) is read
from a declared registry (`schema_registry`), the schema registered under the tenant's subject -
complete example in [Avro payloads under schema, schema registry](https://flexway.io/en/docs/operations-examples-brokers/#avro-payloads-under-schema-schema-registry). An **ordinary Kafka
producer** (a Confluent serializer, no CloudEvents header) is read on a channel **bound** to the tenant
or **public** whose only message declares that format: the gateway builds the envelope (`id` =
`<topic>/<partition>/<offset>`, Kafka timestamp), the tenant coming from the bound channel alone.

**Native Kafka clients** (`1.6.0`): beyond an HTTP caller, a Kafka client of any
library connects to the **Kafka listener** (`listeners.kafka`); an operation marked
`x-flexway-kafka-proxy` serves it the channel as a topic of a virtual cluster, its names written in its
tenant's namespace, under the identities of HTTP (a token over OAUTHBEARER, an API key over PLAIN, a
client certificate) - complete example in [Native Kafka clients: a virtual cluster per tenant](https://flexway.io/en/docs/operations-kafka-clients/).

**Every broker, every case** - one complete, validated example per broker in [Examples: Kafka, RabbitMQ, MQTT, Avro, NATS, AMQP 1.0](https://flexway.io/en/docs/operations-examples-brokers/),
NATS and AMQP 1.0 (`1.6.0`) included, files of `examples/` validated like the others. "As" points to the same declaration, which does not depend on the broker.

| Case | Kafka | AMQP 0-9-1 | MQTT | NATS (`1.6.0`) | AMQP 1.0 (`1.6.0`) |
| --- | --- | --- | --- | --- | --- |
| Publish from HTTP | [Kafka](https://flexway.io/en/docs/operations-examples-brokers/#kafka-publish-stream-deliver) | [RabbitMQ](https://flexway.io/en/docs/operations-examples-brokers/#rabbitmq-in-amqp-0-9-1-a-channel-per-person-a-public-feed) | [MQTT 5](https://flexway.io/en/docs/operations-examples-brokers/#mqtt-5-a-claim-the-tenant-attribute-a-webhook) | [JetStream](https://flexway.io/en/docs/operations-examples-brokers/#nats-jetstream-publish-from-http-or-an-mcp-tool) (JetStream; Core refused) | [Service Bus](https://flexway.io/en/docs/operations-examples-brokers/#amqp-10-publish-to-azure-service-bus-deliver-from-a-subscription) |
| Publish through an MCP tool | [Kafka](https://flexway.io/en/docs/operations-examples-brokers/#kafka-publish-stream-deliver) | as [Kafka](https://flexway.io/en/docs/operations-examples-brokers/#kafka-publish-stream-deliver) | as [Kafka](https://flexway.io/en/docs/operations-examples-brokers/#kafka-publish-stream-deliver) | [JetStream](https://flexway.io/en/docs/operations-examples-brokers/#nats-jetstream-publish-from-http-or-an-mcp-tool) | as [JetStream](https://flexway.io/en/docs/operations-examples-brokers/#nats-jetstream-publish-from-http-or-an-mcp-tool) |
| SSE fan-out, `Last-Event-ID` resumption | [Kafka](https://flexway.io/en/docs/operations-examples-brokers/#kafka-publish-stream-deliver) | [RabbitMQ](https://flexway.io/en/docs/operations-examples-brokers/#rabbitmq-in-amqp-0-9-1-a-channel-per-person-a-public-feed) | [MQTT 5](https://flexway.io/en/docs/operations-examples-brokers/#mqtt-5-a-claim-the-tenant-attribute-a-webhook) (no resumption) | [NATS](https://flexway.io/en/docs/operations-examples-brokers/#nats-sse-and-websocket-bound-to-the-tenant-wildcards-nats-core) | [RabbitMQ 4](https://flexway.io/en/docs/operations-examples-brokers/#amqp-10-consume-rabbitmq-4-in-sse-and-websocket) |
| Acknowledged WebSocket | [Kafka](https://flexway.io/en/docs/operations-examples-brokers/#kafka-publish-stream-deliver) | [RabbitMQ](https://flexway.io/en/docs/operations-examples-brokers/#rabbitmq-in-amqp-0-9-1-a-channel-per-person-a-public-feed) | as [Kafka](https://flexway.io/en/docs/operations-examples-brokers/#kafka-publish-stream-deliver) | [NATS](https://flexway.io/en/docs/operations-examples-brokers/#nats-sse-and-websocket-bound-to-the-tenant-wildcards-nats-core) (JetStream only) | [RabbitMQ 4](https://flexway.io/en/docs/operations-examples-brokers/#amqp-10-consume-rabbitmq-4-in-sse-and-websocket) (RabbitMQ 4, Artemis, Solace) |
| Webhook and dead-letter channel | [Kafka](https://flexway.io/en/docs/operations-examples-brokers/#kafka-publish-stream-deliver) | as [Kafka](https://flexway.io/en/docs/operations-examples-brokers/#kafka-publish-stream-deliver) | [MQTT 5](https://flexway.io/en/docs/operations-examples-brokers/#mqtt-5-a-claim-the-tenant-attribute-a-webhook) (MQTT 5 only) | [JetStream webhook](https://flexway.io/en/docs/operations-examples-brokers/#nats-jetstream-a-webhook-from-a-durable-consumer) (JetStream only) | [Service Bus](https://flexway.io/en/docs/operations-examples-brokers/#amqp-10-publish-to-azure-service-bus-deliver-from-a-subscription) |
| Bound to the tenant | [Kafka](https://flexway.io/en/docs/operations-examples-brokers/#kafka-publish-stream-deliver) | as [Kafka](https://flexway.io/en/docs/operations-examples-brokers/#kafka-publish-stream-deliver) | [MQTT 5](https://flexway.io/en/docs/operations-examples-brokers/#mqtt-5-a-claim-the-tenant-attribute-a-webhook) (`tenant` attribute) | [JetStream](https://flexway.io/en/docs/operations-examples-brokers/#nats-jetstream-publish-from-http-or-an-mcp-tool), [NATS](https://flexway.io/en/docs/operations-examples-brokers/#nats-sse-and-websocket-bound-to-the-tenant-wildcards-nats-core) | [Service Bus](https://flexway.io/en/docs/operations-examples-brokers/#amqp-10-publish-to-azure-service-bus-deliver-from-a-subscription) |
| Bound to the subject | as [RabbitMQ](https://flexway.io/en/docs/operations-examples-brokers/#rabbitmq-in-amqp-0-9-1-a-channel-per-person-a-public-feed) | [RabbitMQ](https://flexway.io/en/docs/operations-examples-brokers/#rabbitmq-in-amqp-0-9-1-a-channel-per-person-a-public-feed) | as [RabbitMQ](https://flexway.io/en/docs/operations-examples-brokers/#rabbitmq-in-amqp-0-9-1-a-channel-per-person-a-public-feed) | as [RabbitMQ](https://flexway.io/en/docs/operations-examples-brokers/#rabbitmq-in-amqp-0-9-1-a-channel-per-person-a-public-feed) | [RabbitMQ 4](https://flexway.io/en/docs/operations-examples-brokers/#amqp-10-consume-rabbitmq-4-in-sse-and-websocket) |
| Bound to a claim | as [MQTT 5](https://flexway.io/en/docs/operations-examples-brokers/#mqtt-5-a-claim-the-tenant-attribute-a-webhook) | as [MQTT 5](https://flexway.io/en/docs/operations-examples-brokers/#mqtt-5-a-claim-the-tenant-attribute-a-webhook) | [MQTT 5](https://flexway.io/en/docs/operations-examples-brokers/#mqtt-5-a-claim-the-tenant-attribute-a-webhook) | [NATS](https://flexway.io/en/docs/operations-examples-brokers/#nats-sse-and-websocket-bound-to-the-tenant-wildcards-nats-core) | as [NATS](https://flexway.io/en/docs/operations-examples-brokers/#nats-sse-and-websocket-bound-to-the-tenant-wildcards-nats-core) |
| Public channel | as [RabbitMQ](https://flexway.io/en/docs/operations-examples-brokers/#rabbitmq-in-amqp-0-9-1-a-channel-per-person-a-public-feed) | [RabbitMQ](https://flexway.io/en/docs/operations-examples-brokers/#rabbitmq-in-amqp-0-9-1-a-channel-per-person-a-public-feed) | as [RabbitMQ](https://flexway.io/en/docs/operations-examples-brokers/#rabbitmq-in-amqp-0-9-1-a-channel-per-person-a-public-feed) | [NATS](https://flexway.io/en/docs/operations-examples-brokers/#nats-sse-and-websocket-bound-to-the-tenant-wildcards-nats-core) | [RabbitMQ 4](https://flexway.io/en/docs/operations-examples-brokers/#amqp-10-consume-rabbitmq-4-in-sse-and-websocket) |
| Structured CloudEvents | as [RabbitMQ](https://flexway.io/en/docs/operations-examples-brokers/#rabbitmq-in-amqp-0-9-1-a-channel-per-person-a-public-feed) | [RabbitMQ](https://flexway.io/en/docs/operations-examples-brokers/#rabbitmq-in-amqp-0-9-1-a-channel-per-person-a-public-feed) | as [RabbitMQ](https://flexway.io/en/docs/operations-examples-brokers/#rabbitmq-in-amqp-0-9-1-a-channel-per-person-a-public-feed) (always on 3.1.1) | [NATS](https://flexway.io/en/docs/operations-examples-brokers/#nats-sse-and-websocket-bound-to-the-tenant-wildcards-nats-core) | [RabbitMQ 4](https://flexway.io/en/docs/operations-examples-brokers/#amqp-10-consume-rabbitmq-4-in-sse-and-websocket) |
| The gateway's identity | [Kafka](https://flexway.io/en/docs/operations-examples-brokers/#kafka-publish-stream-deliver): SCRAM over TLS, mTLS | [RabbitMQ](https://flexway.io/en/docs/operations-examples-brokers/#rabbitmq-in-amqp-0-9-1-a-channel-per-person-a-public-feed): PLAIN, virtual host | [MQTT 5](https://flexway.io/en/docs/operations-examples-brokers/#mqtt-5-a-claim-the-tenant-attribute-a-webhook): PLAIN over `mqtts://` | [JetStream](https://flexway.io/en/docs/operations-examples-brokers/#nats-jetstream-publish-from-http-or-an-mcp-tool) credentials, [NATS](https://flexway.io/en/docs/operations-examples-brokers/#nats-sse-and-websocket-bound-to-the-tenant-wildcards-nats-core) token and password, [JetStream webhook](https://flexway.io/en/docs/operations-examples-brokers/#nats-jetstream-a-webhook-from-a-durable-consumer) nkey | [Service Bus](https://flexway.io/en/docs/operations-examples-brokers/#amqp-10-publish-to-azure-service-bus-deliver-from-a-subscription): SAS key, [RabbitMQ 4](https://flexway.io/en/docs/operations-examples-brokers/#amqp-10-consume-rabbitmq-4-in-sse-and-websocket): PLAIN, EXTERNAL with a certificate |