Flexway - Publishing and consuming on a broker (Operations)

Operations guide: configure, deploy and observe the gateway. What the capability does, product side: product guide; calling it: developer guide.

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.

# 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 }

# 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 }

# 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:

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

Every broker, every case - one complete, validated example per broker in Examples: Kafka, RabbitMQ, MQTT, Avro, NATS, AMQP 1.0, 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	RabbitMQ	MQTT 5	JetStream (JetStream; Core refused)	Service Bus
Publish through an MCP tool	Kafka	as Kafka	as Kafka	JetStream	as JetStream
SSE fan-out, `Last-Event-ID` resumption	Kafka	RabbitMQ	MQTT 5 (no resumption)	NATS	RabbitMQ 4
Acknowledged WebSocket	Kafka	RabbitMQ	as Kafka	NATS (JetStream only)	RabbitMQ 4 (RabbitMQ 4, Artemis, Solace)
Webhook and dead-letter channel	Kafka	as Kafka	MQTT 5 (MQTT 5 only)	JetStream webhook (JetStream only)	Service Bus
Bound to the tenant	Kafka	as Kafka	MQTT 5 (`tenant` attribute)	JetStream, NATS	Service Bus
Bound to the subject	as RabbitMQ	RabbitMQ	as RabbitMQ	as RabbitMQ	RabbitMQ 4
Bound to a claim	as MQTT 5	as MQTT 5	MQTT 5	NATS	as NATS
Public channel	as RabbitMQ	RabbitMQ	as RabbitMQ	NATS	RabbitMQ 4
Structured CloudEvents	as RabbitMQ	RabbitMQ	as RabbitMQ (always on 3.1.1)	NATS	RabbitMQ 4
The gateway's identity	Kafka: SCRAM over TLS, mTLS	RabbitMQ: PLAIN, virtual host	MQTT 5: PLAIN over `mqtts://`	JetStream credentials, NATS token and password, JetStream webhook nkey	Service Bus: SAS key, RabbitMQ 4: PLAIN, EXTERNAL with a certificate

Publishing and consuming on a broker

Next steps