# Exemples : Kafka, RabbitMQ, MQTT, Avro, NATS, AMQP 1.0

> Le parcours complet de Publier et consommer sur un courtier sur Kafka : une porte de publication, aussi outil MCP ; un flux SSE en éventail ou WebSocket…

Exemple - Depuis la version 1.5.0 - Flexway v1.6.3 - https://flexway.io/docs/operations-examples-brokers/

> Guide d'exploitation : configurer, déployer et observer la passerelle. Ce que fait la capacité côté produit : [guide produit](https://flexway.io/docs/product-overview/) ; l'appeler : [guide développeur](https://flexway.io/docs/developer-overview/).

## Kafka : publier, diffuser, livrer

Le parcours complet de [Publier et consommer sur un courtier](https://flexway.io/docs/operations-async/) sur Kafka : une porte de publication, aussi
outil MCP ; un flux SSE en éventail ou WebSocket accusé, lié au tenant ; les évènements d'`acme` livrés en
webhook signé, le rebut dans `orders.dead`. La passerelle s'authentifie en SASL SCRAM sur TLS.

<!-- example: examples/async-kafka/flexway.yaml -->

```yaml
# Kafka behind the gateway's controls (1.5.0): an HTTP door and an MCP tool publish, a
# browser streams its tenant's events in SSE or WebSocket, a partner receives them as signed webhooks.
# flexway.yaml lists no channel - they come from the AsyncAPI contract.
version: 1
contracts:
  openapi: ./openapi.yaml
  asyncapi: ./asyncapi.yaml                      # AsyncAPI 3.0.0 or 3.1.0
upstreams:
  events:
    protocol: kafka
    brokers: [kafka-0.kafka:9093, kafka-1.kafka:9093]
    tls: { server_name: kafka.internal }          # TLS, system roots; ca_file, cert_file and key_file for a private CA or mTLS
    sasl: { mechanism: SCRAM-SHA-512, username: flexway, password: "file:/etc/flexway-secrets/kafka/password" }
    timeout: 2s                                  # one publication's budget, the 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"]
    retries: { max: 8, max_delay: 10m }
auth:
  issuers:
    - name: customers
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: api
      tenant_claim: org                          # the tenant every channel below is bound to
agents:
  mcp:
    resources:
      api.example.com: https://api.example.com/mcp
```

```yaml
# examples/async-kafka/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Orders events, version: 1.0.0 }
channels:
  orderCreated:
    address: "orders.{tenant}.created"           # one topic per tenant
    parameters:
      tenant: { x-flexway-bind: tenant }         # bound to the verified identity, never read from the request
    messages:
      created:
        contentType: application/json
        payload:
          type: object
          required: [orderId]
          properties:
            orderId: { type: string, minLength: 1 }
operations:
  publishOrderCreated:
    action: send
    channel: { $ref: "#/channels/orderCreated" }
    x-flexway-upstream: events
    x-flexway-key: /orderId                      # ordered per key, and per key only
  receiveOrderCreated:
    action: receive
    channel: { $ref: "#/channels/orderCreated" }
    x-flexway-upstream: events
    x-flexway-deliver:                           # acme's events to its webhook; what fails ends in orders.dead
      - { upstream: partner-hook, tenant: acme, dead_letter: orders.dead }
```

```yaml
# examples/async-kafka/openapi.yaml
openapi: 3.1.0
info: { title: Orders events, version: "1.0.0" }
components:
  securitySchemes:
    customers: { type: openIdConnect, openIdConnectUrl: https://login.example.net/.well-known/openid-configuration }
paths:
  /orders/events:
    post:
      operationId: emitOrderCreated
      summary: Publish an order-created event for the caller's tenant
      x-flexway-publish: { operation: publishOrderCreated, idempotency: required }
      x-flexway-tool: { name: emit_order_created }   # the same door, offered to agents as an MCP tool
      security: [ { customers: [orders:write] } ]
      requestBody:
        required: true
        content:
          application/json:
            schema: { type: object, required: [orderId], properties: { orderId: { type: string, minLength: 1 } } }
      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, buffer: 256, history: 64 }
      security: [ { customers: [orders:read] } ]
      responses: { "200": { description: The caller's tenant's events } }
```

| Cas | Ce que fait la passerelle |
| --- | --- |
| `POST /orders/events` ou l'outil `emit_order_created` | `202` après `acks=all` ; `Idempotency-Key` exigée (`428` sinon), devenue l'`id` ; ordre par `orderId` |
| `GET /orders/stream` en SSE | Un consommateur par canal et par réplique ; `Last-Event-ID` repris dans les 64 derniers messages, `412 precondition_failed` au-delà |
| `GET /orders/stream` en WebSocket | Groupe propre à l'abonné, `{"ack": "<id>"}` dans les 30 s, sinon fermeture `1008` et nouvelle livraison |
| Livraison à `partner-hook` | Groupe partagé par les répliques, Standard Webhooks, position validée après le `2xx`, rebut `orders.dead` |
| mTLS au lieu de SASL | `tls.cert_file` et `tls.key_file` montés, sans `sasl` |

## RabbitMQ en AMQP 0-9-1 : canal par personne, flux public

Chaque joueur publie et lit **ses** notifications - le canal est lié au sujet vérifié du jeton - ; le
tableau des scores est un flux **public**, lu sans jeton, repris sur 128 messages. Un producteur hors de la
passerelle y écrit en `application/cloudevents+json` : lu en mode structuré. La passerelle se connecte avec
son utilisateur, sur son hôte virtuel (`/games`, le chemin de l'URL).

<!-- example: examples/async-amqp/flexway.yaml -->

```yaml
# RabbitMQ in AMQP 0-9-1 behind the gateway's controls (1.5.0): each player reads its
# own notifications only (bound to the verified subject), anyone - anonymous included - reads the
# declared public feed. The gateway authenticates itself: its own user, its own virtual host.
version: 1
contracts:
  openapi: ./openapi.yaml
  asyncapi: ./asyncapi.yaml
upstreams:
  bus:
    protocol: amqp                               # AMQP 0-9-1 (protocol_version: 0-9-1, the default)
    brokers: ["amqps://rabbitmq.internal:5671/games"]   # the path is the virtual host; never a password in the URL
    sasl: { mechanism: PLAIN, username: flexway, password: "file:/etc/flexway-secrets/rabbitmq/password" }
    timeout: 2s                                  # publisher confirm included
auth:
  issuers:
    - name: players
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: games
      tenant_claim: org
```

```yaml
# examples/async-amqp/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Game events, version: 1.0.0 }
channels:
  notifications:
    # A topic exchange the broker already has: the gateway never declares one.
    address: "players.{sub}.notifications"
    parameters:
      sub: { x-flexway-bind: subject }           # the verified subject: a player reads its own channel only
    bindings:
      amqp: { is: routingKey, exchange: { name: notifications, type: topic } }
    messages:
      note:
        contentType: application/json
        payload: { type: object, required: [text], properties: { text: { type: string } } }
  scoreboard:
    address: scores.public                       # binds nothing, filters nothing: declared public below
    bindings:
      amqp: { is: routingKey, exchange: { name: scores, type: topic } }
    messages:
      # Written by a producer outside the gateway (an outbox) as application/cloudevents+json: read in
      # structured mode; any other content type is read in binary mode.
      score:
        payload: { type: object, required: [match], properties: { match: { type: string }, score: { type: string } } }
operations:
  publishNotification: { action: send, channel: { $ref: "#/channels/notifications" }, x-flexway-upstream: bus }
  receiveNotifications: { action: receive, channel: { $ref: "#/channels/notifications" }, x-flexway-upstream: bus }
  receiveScores:
    action: receive
    channel: { $ref: "#/channels/scoreboard" }
    x-flexway-upstream: bus
    x-flexway-public: true                       # read by anyone, anonymous included
```

```yaml
# examples/async-amqp/openapi.yaml
openapi: 3.1.0
info: { title: Game events, version: "1.0.0" }
components:
  securitySchemes:
    players: { type: openIdConnect, openIdConnectUrl: https://login.example.net/.well-known/openid-configuration }
paths:
  /me/notifications:
    post:
      operationId: notifyMe
      x-flexway-publish: { operation: publishNotification }
      security: [ { players: [] } ]
      requestBody:
        required: true
        content:
          application/json:
            schema: { type: object, required: [text], properties: { text: { type: string } } }
      responses: { "202": { description: Confirmed by the broker } }
    get:
      operationId: streamMyNotifications
      # WebSocket: at least once - the client acknowledges {"ack": "<id>"} within ack_timeout.
      x-flexway-subscribe: { operation: receiveNotifications, transport: [sse, websocket], max_streams: 3, ack_timeout: 20s }
      security: [ { players: [] } ]
      responses: { "200": { description: The caller's own notifications } }
  /scores:
    get:
      operationId: streamScores
      # SSE only; one consumer per replica whatever the audience, 128 messages kept for Last-Event-ID.
      x-flexway-subscribe: { operation: receiveScores, transport: [sse], max_streams: 2, max_anonymous_streams: 500, history: 128 }
      security: []                               # anonymous door: only toward a public operation
      responses: { "200": { description: The public scoreboard } }
```

| Cas | Ce que fait la passerelle |
| --- | --- |
| `POST /me/notifications` | Publié sur `players.<sub>.notifications` de l'échange `notifications`, `202` après la confirmation |
| `GET /me/notifications` | Le seul canal du sujet ; une valeur hors des lettres, chiffres, `_` et `-` est refusée (`403`) |
| `GET /scores` sans jeton | Public déclaré : jusqu'à 500 anonymes ensemble, 2 flux par adresse ; un abonné trop lent fermé, les autres continuent |
| Webhook | Même déclaration qu'en [Kafka : publier, diffuser, livrer](#kafka--publier-diffuser-livrer) : une file nommée partagée par les répliques |

## MQTT 5 : claim, attribut tenant, webhook

Un opérateur lit la télémétrie de **son** site - le canal est lié à la claim `site` du jeton -, sans reprise
(`history: 0`) ; les alertes de tous les tenants passent par **un** sujet commun, chacune portant l'attribut
`tenant` : la consommation est filtrée sur lui, et les alertes d'`acme` partent en webhook par un abonnement
**partagé** MQTT 5 (une réplique par message).

<!-- example: examples/async-mqtt/flexway.yaml -->

```yaml
# An MQTT 5 broker behind the gateway's controls (1.5.0): an operator's screen reads
# the telemetry of its own site (bound to a claim of its token), alerts carry the tenant attribute
# and are delivered to the tenant's webhook through a shared subscription - one replica per message.
version: 1
contracts:
  openapi: ./openapi.yaml
  asyncapi: ./asyncapi.yaml
upstreams:
  plant:
    protocol: mqtt
    protocol_version: "5"                        # the default; 3.1.1 has no shared subscription, so no webhook
    brokers: ["mqtts://mqtt.internal:8883"]
    sasl: { mechanism: PLAIN, username: flexway, password: "file:/etc/flexway-secrets/mqtt/password" }
    timeout: 2s                                  # PUBACK at QoS 1 included
  ops-hook:
    protocol: webhook
    url: https://ops.acme.example/flexway/alerts
    secrets: ["file:/etc/flexway-secrets/hooks/ops"]
    retries: { max: 5, max_delay: 5m }
auth:
  issuers:
    - name: operators
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: plant
      tenant_claim: org
```

```yaml
# examples/async-mqtt/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Plant telemetry, version: 1.0.0 }
channels:
  siteTelemetry:
    address: "sites/{site}/telemetry"
    parameters:
      site: { x-flexway-bind: "claim:site" }     # a claim of the verified token: an operator reads its own site
    messages:
      reading:
        payload: { type: object, required: [sensor, value], properties: { sensor: { type: string }, value: { type: number } } }
  alerts:
    address: plant/alerts                        # one topic for every tenant ...
    messages:
      alert:
        headers:                                 # ... kept apart by the tenant attribute every message declares
          type: object
          properties: { tenant: { type: string } }
        payload: { type: object, required: [code], properties: { code: { type: string } } }
operations:
  publishReading: { action: send, channel: { $ref: "#/channels/siteTelemetry" }, x-flexway-upstream: plant }
  receiveReadings: { action: receive, channel: { $ref: "#/channels/siteTelemetry" }, x-flexway-upstream: plant }
  publishAlert: { action: send, channel: { $ref: "#/channels/alerts" }, x-flexway-upstream: plant }
  receiveAlerts:
    action: receive
    channel: { $ref: "#/channels/alerts" }
    x-flexway-upstream: plant
    x-flexway-deliver:                           # acme's alerts only; what cannot be delivered ends in plant/alerts-dead
      - { upstream: ops-hook, tenant: acme, dead_letter: plant/alerts-dead }
```

```yaml
# examples/async-mqtt/openapi.yaml
openapi: 3.1.0
info: { title: Plant telemetry, version: "1.0.0" }
components:
  securitySchemes:
    operators: { type: openIdConnect, openIdConnectUrl: https://login.example.net/.well-known/openid-configuration }
paths:
  /telemetry:
    post:
      operationId: postReading
      x-flexway-publish: { operation: publishReading, idempotency: required }
      security: [ { operators: [telemetry:write] } ]
      requestBody:
        required: true
        content:
          application/json:
            schema: { type: object, required: [sensor, value], properties: { sensor: { type: string }, value: { type: number } } }
      responses: { "202": { description: Acknowledged by the broker (PUBACK) } }
    get:
      operationId: streamReadings
      x-flexway-subscribe: { operation: receiveReadings, transport: [sse], history: 0 }   # no resumption
      security: [ { operators: [telemetry:read] } ]
      responses: { "200": { description: The telemetry of the caller's site } }
  /alerts:
    post:
      operationId: raiseAlert
      x-flexway-publish: { operation: publishAlert }
      security: [ { operators: [alerts:write] } ]
      requestBody:
        required: true
        content:
          application/json:
            schema: { type: object, required: [code], properties: { code: { type: string } } }
      responses: { "202": { description: Acknowledged by the broker (PUBACK) } }
```

| Cas | Ce que fait la passerelle |
| --- | --- |
| `POST /telemetry` | `202` après le `PUBACK` QoS 1 ; `Idempotency-Key` exigée |
| `GET /telemetry` | Le canal `sites/<site>/telemetry` du jeton seulement ; sans claim `site`, refusé |
| `POST /alerts` | L'attribut `tenant` vient de l'identité, jamais du corps |
| Livraison à `ops-hook` | Abonnement partagé, filtre `tenant` = `acme`, rebut `plant/alerts-dead` ; refusée au chargement en MQTT 3.1.1 |
| mTLS au lieu du mot de passe | `tls.cert_file` et `tls.key_file` montés, sans `sasl` |

## Charges Avro sous schéma, registre de schémas

Une boutique publie ses commandes en **Avro** depuis HTTP : le corps est décodé sous `order.avsc`,
compilé au chargement, avant d'atteindre Kafka ; un tableau de bord les lit en SSE **en JSON**, un
service les lit en WebSocket **en Avro tel quel**. Le canal d'audit porte le **format filaire de
Confluent** : chaque identifiant de schéma est lu dans le registre, et doit y être enregistré sous le
sujet du tenant (`<tenant>.orders.audit-value`) - un autre tenant ne publie jamais sous ce schéma.

<!-- example: examples/async-binary/flexway.yaml -->

```yaml
# Avro orders through the asynchronous plane (1.6.0): an HTTP caller publishes an Avro
# body - checked under its schema, compiled at load -, a subscriber over SSE or a webhook reads it in
# JSON, a WebSocket that declares binary frames reads it as it is. The audit channel carries Confluent's
# wire format: each schema id is read in the registry and must be registered under the tenant's subject.
version: 1
contracts:
  openapi: ./openapi.yaml
  asyncapi: ./asyncapi.yaml
upstreams:
  orders-brokers:
    protocol: kafka
    brokers: [kafka-0.internal:9092, kafka-1.internal:9092]
schema_registry:
  url: https://registry.internal:8081                # read, never written - the Confluent API
  auth:
    username: flexway
    password: ${REGISTRY_PASSWORD}                    # a reference, resolved at each request
  timeout: 2s                                         # a lookup holds the message that needs it
  max_schemas: 1000                                   # compiled once each, kept by id
auth:
  issuers:
    - name: customers
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: api
      tenant_claim: org
```

```yaml
# examples/async-binary/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Shop orders, version: "1.0.0" }
channels:
  orders:
    address: "{tenant}.orders"
    parameters:
      tenant: { x-flexway-bind: tenant }
    messages:
      order:
        contentType: application/vnd.apache.avro+binary
        payload: { schemaFormat: "application/vnd.apache.avro;version=1.9.0", schema: { $ref: "./order.avsc" } }
  audit:
    address: "{tenant}.orders.audit"
    parameters:
      tenant: { x-flexway-bind: tenant }
    messages:
      audited:
        contentType: application/vnd.apache.avro+binary
        payload: { schemaFormat: "application/vnd.apache.avro;version=1.9.0", schema: { $ref: "./order.avsc" } }
        bindings:
          kafka: { schemaIdLocation: payload, schemaIdPayloadEncoding: confluent, schemaLookupStrategy: TopicIdStrategy }
operations:
  publishOrder:
    action: send
    channel: { $ref: "#/channels/orders" }
    x-flexway-upstream: orders-brokers
    x-flexway-validation: full          # decode every body under order.avsc - opt-in for a binary message
    x-flexway-key: /orderId             # read in the payload's JSON form
  receiveOrder:
    action: receive
    channel: { $ref: "#/channels/orders" }
    x-flexway-upstream: orders-brokers
  publishAudit:
    action: send
    channel: { $ref: "#/channels/audit" }
    x-flexway-upstream: orders-brokers
    x-flexway-validation: full          # the schema id in the registry, under <tenant>.orders.audit-value
  receiveAudit:
    # Records of an ordinary Kafka producer (a Confluent serializer, no CloudEvents header) are read
    # too: the channel is bound to the tenant and its only message is in the registry's wire format,
    # so the gateway builds each envelope - id <topic>/<partition>/<offset>, the record's timestamp.
    action: receive
    channel: { $ref: "#/channels/audit" }
    x-flexway-upstream: orders-brokers
```

```yaml
# examples/async-binary/openapi.yaml
openapi: 3.1.0
info: { title: Shop orders, version: "1.0.0" }
components:
  securitySchemes:
    customers: { type: openIdConnect, openIdConnectUrl: https://login.example.net/.well-known/openid-configuration }
security: [{ customers: [] }]
paths:
  /orders/events:
    post:
      operationId: emitOrder
      x-flexway-publish: { operation: publishOrder, idempotency: required }
      requestBody:
        required: true
        content: { application/vnd.apache.avro+binary: { schema: { type: string, format: binary } } }
      responses: { "202": { description: published, after the broker's acknowledgement } }
  /orders/audit:
    post:
      operationId: emitAudit
      x-flexway-publish: { operation: publishAudit }
      requestBody:
        required: true
        content: { application/vnd.apache.avro+binary: { schema: { type: string, format: binary } } }
      responses: { "202": { description: published } }
  /orders/stream:
    get:
      operationId: streamOrders
      x-flexway-subscribe: { operation: receiveOrder, transport: [sse] }
      responses: { "200": { description: the tenant's orders, in JSON } }
  /orders/frames:
    get:
      operationId: framesOrders
      x-flexway-subscribe: { operation: receiveOrder, transport: [websocket], frames: binary }
      responses: { "200": { description: the tenant's orders, the Avro payload in binary frames } }
  /orders/audit/stream:
    get:
      operationId: streamAudit
      x-flexway-subscribe: { operation: receiveAudit, transport: [sse] }
      responses: { "200": { description: the tenant's audit records in JSON, ordinary producers' included } }
```

| Cas | Ce que fait la passerelle |
| --- | --- |
| `POST /orders/events`, corps Avro conforme | `202` après l'accusé ; clé Kafka lue en `/orderId` de la forme JSON du message |
| Corps qui ne se décode pas sous `order.avsc` | `422 invalid_request`, la règle seule, jamais une valeur du message |
| Autre `Content-Type` | `422 invalid_request` (la porte ne déclare que l'Avro) ; une porte sans `requestBody` répond `415 unsupported_media_type` |
| `POST /orders/audit`, identifiant enregistré sous `acme.orders.audit-value` | `202` ; schéma lu une fois par identifiant, gardé en mémoire (`max_schemas`) |
| Identifiant inconnu, ou enregistré sous un autre sujet | `422 invalid_request` |
| Registre injoignable ou en erreur | `503 upstream_unavailable` avec `Retry-After` : jamais un message publié sans validation ; coupe-circuit propre au registre |
| `GET /orders/stream` (SSE) | Chaque évènement en JSON (encodage JSON d'Avro), `datacontenttype: application/json` ; un message qui ne se décode pas n'est pas remis (`flexway_async_deliver_total{outcome="invalid"}`) |
| `GET /orders/frames` (WebSocket) | Une trame texte (les attributs, dont l'`id` à accuser), puis une trame **binaire** : la charge Avro telle quelle |
| Webhook (`x-flexway-deliver`) | Corps CloudEvent en JSON transcodé ; registre injoignable : le message n'est ni livré ni accusé, il revient |
| Protobuf | `schemaFormat: "application/vnd.google.protobuf;version=3"` et `schema: { $ref: "./reading.binpb#acme.Reading" }` - un jeu de descripteurs compilé (`protoc --descriptor_set_out --include_imports`, `buf build`), le message par son nom ; une source `.proto` en ligne est refusée |
| CBOR | `contentType: application/cbor` et un schéma JSON : la charge est lue dans le modèle JSON de CBOR (RFC 8949 §6.1) et vérifiée contre ce schéma |
| `GET /orders/audit/stream`, enregistrement d'un **producteur Kafka ordinaire** (format filaire, aucun en-tête CloudEvents) | Remis dans l'enveloppe que construit la passerelle : `id` = `acme.orders.audit/<partition>/<position>` (stable d'une réplique à l'autre, c'est l'`id` SSE et le `Last-Event-ID`, l'`id` à accuser en WebSocket, le `webhook-id`), `source` = `/flexway/orders-brokers/audit`, `type` = `audited`, `time` = l'horodatage Kafka, `datacontenttype: application/json` après transcodage ; **aucun** attribut `tenant` : le tenant est celui du canal lié, jamais un en-tête ni le contenu du message |
| Enregistrement sans attribut sur un autre canal, ou qui porte un attribut `ce_*` sans en être un | Pas un évènement, comme en `1.5.0` : non remis (`flexway_async_deliver_total{outcome="invalid"}`) |
| Schéma d'écriture du registre différent du schéma du contrat | **Résolu** vers le schéma du contrat par les règles standard d'Avro (champ ajouté avec défaut, champ retiré, `int` -> `long`, alias, symbole d'énumération par défaut) : même JSON qu'un message écrit sous le schéma du contrat ; résolution une fois par paire (identifiant, schéma du contrat), `max_schemas` paires gardées. Schémas identiques (même forme canonique) : aucun coût de plus |
| Schéma d'écriture qui ne se résout pas vers celui du contrat (champ sans défaut, type incompatible) | Refusé entier, jamais livré en partie : `422 invalid_request` à la publication ; à la consommation, non remis comme une charge qui ne se décode pas - SSE au plus une fois, WebSocket accusé, webhook mis au rebut -, la consommation continue ; compté `outcome="incompatible"` |
| `x-flexway-key` sur un message binaire | La clé est lue dans la forme **JSON** du message : **un décodage par publication**, même avec `x-flexway-validation: off` (une charge qui ne se décode pas est alors refusée en `422`) - un coût à peser sur un débit élevé ; sans clé, une charge non validée est relayée sans être lue |

**Refusé au chargement** : un `schemaFormat` inconnu (RAML, OpenAPI, Avro `2.x`), un schéma Avro
invalide, récursif ou dont un tableau contient des éléments sans octet, un message absent du jeu de
descripteurs, un `$ref` de schéma binaire hors du répertoire du document AsyncAPI, le format filaire
sans `schema_registry` ou sur un courtier qui n'est pas Kafka, `x-flexway-validation: off` sur un
message JSON (toujours vérifié), `frames: binary` sans WebSocket ou sur un message JSON, une
consommation d'un canal au format filaire isolée par le seul attribut `tenant` (ni liée, ni publique) :
un producteur ordinaire n'écrit aucun attribut, et le tenant n'est jamais lu dans un message.

**Mesurer** : `flexway_async_payloads_total{operation, format, step, outcome}` - `step` vaut `validate`
ou `transcode`, `outcome` `ok`, `invalid`, `not_attached`, `incompatible` ou `unavailable` ; jamais une valeur du
message ni un identifiant de schéma.

## NATS JetStream : publier depuis HTTP ou un outil MCP

> Disponible depuis `1.6.0` ; exemple validé par le binaire et épinglé par les tests.

Un service HTTP ou un agent publie dans un flux JetStream **sans client NATS** ; `202` seulement après
le `PubAck` du flux, l'`Idempotency-Key` devenue `Nats-Msg-Id` - le flux dédoublonne un renvoi après un `504`.
La passerelle ne crée **aucun** flux : le sujet doit être capturé par un flux existant, sinon `502 upstream_error`.

<!-- example: examples/async-nats-jetstream/flexway.yaml -->

```yaml
# NATS JetStream behind the gateway's controls (1.6.0): an HTTP door and an MCP tool publish
# to a stream without a NATS client - 202 only after the stream's PubAck, the Idempotency-Key as
# Nats-Msg-Id. The gateway creates no stream: a subject no stream captures answers 502.
version: 1
contracts:
  openapi: ./openapi.yaml
  asyncapi: ./asyncapi.yaml
upstreams:
  orders-nats:
    protocol: nats
    brokers: ["tls://nats-0.nats:4222", "tls://nats-1.nats:4222"]   # nats:// or tls://, never a credential in the URL
    tls: { server_name: nats.internal }
    nats:
      credentials: "file:/etc/flexway-secrets/nats/flexway.creds"   # the gateway's NATS user: JWT and nkey seed
      jetstream: true                            # 202 after the stream's PubAck; false (NATS Core) refuses a publication
    timeout: 2s                                  # one publication's budget, the PubAck included
    max_in_flight: 512                           # bounded buffer: past it, 503 overloaded
auth:
  issuers:
    - name: customers
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: api
      tenant_claim: org                          # the tenant every subject below is bound to
agents:
  mcp:
    resources:
      api.example.com: https://api.example.com/mcp
```

```yaml
# examples/async-nats-jetstream/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Orders, version: 1.0.0 }
channels:
  orderCreated:
    address: "orders.{tenant}.created"          # captured by the ORDERS stream (orders.>): the gateway creates no stream
    parameters:
      tenant: { x-flexway-bind: tenant }
    messages:
      created:
        contentType: application/json
        payload: { type: object, required: [orderId], properties: { orderId: { type: string } } }
operations:
  publishOrderCreated:
    action: send
    channel: { $ref: "#/channels/orderCreated" }
    x-flexway-upstream: orders-nats
```

```yaml
# examples/async-nats-jetstream/openapi.yaml
openapi: 3.1.0
info: { title: Orders, version: "1.0.0" }
components:
  securitySchemes:
    customers: { type: openIdConnect, openIdConnectUrl: https://login.example.net/.well-known/openid-configuration }
paths:
  /orders/events:
    post:
      operationId: emitOrderCreated
      x-flexway-publish: { operation: publishOrderCreated, idempotency: required }   # Idempotency-Key -> Nats-Msg-Id
      x-flexway-tool: { name: emit_order_created }   # the same door, offered to agents as an MCP tool
      security: [ { customers: [orders:write] } ]
      requestBody:
        required: true
        content:
          application/json: { schema: { type: object, required: [orderId], properties: { orderId: { type: string } } } }
      responses: { "202": { description: Acknowledged by the stream } }
```

| Ce qui se passe | NATS JetStream |
| --- | --- |
| `POST /orders/events` | Admission, corps vérifié, CloudEvents en mode binaire (en-têtes `ce-*`), publication sur `orders.acme.created`, `202` après le `PubAck` |
| Le même appel en outil MCP (`emit_order_created`) | Même décision, même accusé |
| `jetstream: false` (NATS Core) | **Refusé au chargement** : Core n'accuse rien, un `202` mentirait |
| Aucun flux ne capture le sujet | `502 upstream_error` (« no responders »), jamais un `202` |

## NATS : SSE et WebSocket liés au tenant, jokers, NATS Core

> Disponible depuis `1.6.0` ; exemple validé par le binaire et épinglé par les tests.

Un tableau de bord lit **tous** les évènements de commande de son tenant d'un seul flux, grâce à un joker
**inclus dans le canal** après le segment lié ; un opérateur lit les mesures de **son** site sur un NATS Core
(SSE seul, au plus une fois) ; un écran public lit l'état sans jeton. Chaque message reçu est comparé au motif du canal,
segment par segment : un courtier qui en enverrait trop ne fait jamais fuir un tenant (RG-70).

<!-- example: examples/async-nats-streams/flexway.yaml -->

```yaml
# NATS in server-sent events and WebSocket (1.6.0): a dashboard reads every order event of
# its tenant through a wildcard written in the channel after the bound segment; an operator reads its
# own site's readings from NATS Core (server-sent events only, at most once); a public status is read
# without a token. Two brokers, two identities.
version: 1
contracts:
  openapi: ./openapi.yaml
  asyncapi: ./asyncapi.yaml
upstreams:
  orders-nats:
    protocol: nats
    brokers: ["tls://nats-0.nats:4222"]
    tls: { server_name: nats.internal }
    nats:
      token: "file:/etc/flexway-secrets/nats/token"   # a token, by reference - over TLS only
      jetstream: true
  metrics-core:
    protocol: nats
    brokers: ["nats://nats-edge.plant:4222"]
    sasl: { mechanism: PLAIN, username: flexway, password: "file:/etc/flexway-secrets/nats-edge/password" }
    tls: { server_name: nats-edge.plant }
    nats:
      jetstream: false                           # NATS Core: server-sent events only - no publication, WebSocket nor webhook
auth:
  issuers:
    - name: operators
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: api
      tenant_claim: org
```

```yaml
# examples/async-nats-streams/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Orders and metrics, version: 1.0.0 }
channels:
  orderFeed:
    address: "orders.{tenant}.>"                # the wildcard IN the channel, after the segment bound to the tenant
    parameters:
      tenant: { x-flexway-bind: tenant }
    messages:
      any:
        payload: { type: object, required: [orderId], properties: { orderId: { type: string } } }
  siteMetrics:
    address: "sites.{site}.metrics.*"           # bound to a claim of the token: an operator reads its own site
    parameters:
      site: { x-flexway-bind: "claim:site" }
    messages:
      # Written as application/cloudevents+json by a producer outside the gateway: read in structured mode.
      reading:
        payload: { type: object, required: [value], properties: { value: { type: number } } }
  status:
    address: status.public                       # binds nothing, filters nothing: declared public below
    messages:
      state:
        payload: { type: object, required: [state], properties: { state: { type: string } } }
operations:
  receiveOrders: { action: receive, channel: { $ref: "#/channels/orderFeed" }, x-flexway-upstream: orders-nats }
  receiveMetrics: { action: receive, channel: { $ref: "#/channels/siteMetrics" }, x-flexway-upstream: metrics-core }
  receiveStatus:
    action: receive
    channel: { $ref: "#/channels/status" }
    x-flexway-upstream: orders-nats
    x-flexway-public: true                       # read by anyone, anonymous included
```

```yaml
# examples/async-nats-streams/openapi.yaml
openapi: 3.1.0
info: { title: Orders and metrics, version: "1.0.0" }
components:
  securitySchemes:
    operators: { type: openIdConnect, openIdConnectUrl: https://login.example.net/.well-known/openid-configuration }
paths:
  /orders/stream:
    get:
      operationId: streamOrders
      # SSE: one ordered consumer per channel and replica; WebSocket: one durable consumer per subscriber, AckWait = ack_timeout
      x-flexway-subscribe: { operation: receiveOrders, transport: [sse, websocket], max_streams: 2, ack_timeout: 30s, history: 64 }
      security: [ { operators: [orders:read] } ]
      responses: { "200": { description: The caller's tenant's orders } }
  /metrics/stream:
    get:
      operationId: streamMetrics
      x-flexway-subscribe: { operation: receiveMetrics, transport: [sse], history: 0 }   # NATS Core: SSE only
      security: [ { operators: [metrics:read] } ]
      responses: { "200": { description: The readings of the caller's site } }
  /status:
    get:
      operationId: streamStatus
      x-flexway-subscribe: { operation: receiveStatus, transport: [sse], max_anonymous_streams: 1000 }
      security: []
      responses: { "200": { description: The public status } }
```

| Règle | Comportement |
| --- | --- |
| Joker | `*` et `>` seulement dans un canal lu, **après** les segments liés, `>` en dernier - `orders.>.{tenant}` est refusé ; une valeur liée ne contient jamais `.`, `*` ni `>`. Même règle pour `+` et `#` en MQTT, et pour `*` et `#` dans la clé de liaison AMQP 0-9-1 d'un échange (`audit.{tenant}.#`, servi depuis `1.5.0`) |
| SSE | Un consommateur ordonné JetStream (ou un abonnement Core) par canal et par réplique, quel que soit le nombre d'abonnés ; reprise `Last-Event-ID` dans l'historique du canal, `id` = séquence du flux |
| WebSocket | Un consommateur durable par abonné, `AckWait` = `ack_timeout` ; refusé au chargement sur NATS Core |
| CloudEvents | Binaire (`ce-*`) ou structuré selon le type de contenu ; l'attribut `tenant` filtré dans les en-têtes, jamais dans `data` |

## NATS JetStream : webhook depuis un consommateur durable

> Disponible depuis `1.6.0` ; exemple validé par le binaire et épinglé par les tests.

Un partenaire reçoit les factures de **son** tenant en webhook signé, sans accès à NATS : un consommateur
durable **partagé** par les répliques (une réplique par message), accusé après le `2xx`, rebut déclaré au bout
des réessais. Aucune porte HTTP : le contrat OpenAPI n'est pas nécessaire.

<!-- example: examples/async-nats-webhook/flexway.yaml -->

```yaml
# NATS JetStream delivered by webhook (1.6.0): a partner receives its tenant's invoices as
# signed webhooks, without access to NATS - a durable consumer the replicas share (one replica per
# message), acknowledged after the 2xx, the declared dead letter once the retries are spent. No HTTP
# door: no OpenAPI contract. Refused on NATS Core, which redelivers nothing.
version: 1
contracts:
  asyncapi: ./asyncapi.yaml
upstreams:
  billing-nats:
    protocol: nats
    brokers: ["tls://nats-0.nats:4222"]
    tls: { server_name: nats.internal }
    nats:
      nkey_seed: "file:/etc/flexway-secrets/nats/flexway.nk"   # an nkey seed alone, by reference
      jetstream: true                            # a delivery needs JetStream: a durable consumer the replicas share
  partner-hook:
    protocol: webhook
    url: https://hooks.partner.example/invoices
    secrets: ["file:/etc/flexway-secrets/hooks/current"]
    retries: { max: 8, max_delay: 10m }
```

```yaml
# examples/async-nats-webhook/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Invoices, version: 1.0.0 }
channels:
  invoiceIssued:
    address: "invoices.{tenant}.issued"
    parameters:
      tenant: { x-flexway-bind: tenant }       # a delivery has no caller: bound to the tenant only
    messages:
      issued:
        payload: { type: object, required: [invoiceId], properties: { invoiceId: { type: string } } }
operations:
  receiveInvoices:
    action: receive
    channel: { $ref: "#/channels/invoiceIssued" }
    x-flexway-upstream: billing-nats
    x-flexway-deliver:
      - { upstream: partner-hook, tenant: acme, dead_letter: invoices.dead }   # a subject a stream captures
```

Sur NATS Core, la même livraison est **refusée au chargement** : rien ne relivre un message que la cible n'a pas pris.

## AMQP 1.0 : publier vers Azure Service Bus, livrer depuis un abonnement

> Disponible depuis `1.6.0` ; exemple validé par le binaire et épinglé par les tests.

Un service HTTP publie dans le sujet Service Bus de son tenant (`202` après la disposition `accepted`) ;
un partenaire reçoit, en webhook, ce que lit l'abonnement que l'exploitant a créé. Deux canaux, l'un pour
écrire, l'autre pour lire : aucune extension. La passerelle s'authentifie par une **clé SAS** en SASL PLAIN.

<!-- example: examples/async-amqp1-servicebus/flexway.yaml -->

```yaml
# AMQP 1.0 toward Azure Service Bus (1.6.0): an HTTP service publishes to its tenant's
# topic - 202 after the disposition accepted -, a partner receives by webhook what the subscription the
# operator created reads. Two channels, one to write, one to read: no extension. The gateway
# authenticates with a SAS key over SASL PLAIN.
version: 1
contracts:
  openapi: ./openapi.yaml
  asyncapi: ./asyncapi.yaml
upstreams:
  billing-bus:
    protocol: amqp
    protocol_version: "1.0"                      # AMQP 1.0; 0-9-1 stays the default
    brokers: ["amqps://acme-billing.servicebus.windows.net"]
    sasl: { mechanism: PLAIN, username: flexway-send, password: "file:/etc/flexway-secrets/servicebus/key" }   # SAS key name and key
    amqp1: { broker: servicebus }                # publication and webhook; server-sent events and WebSocket refused
    timeout: 3s                                  # the disposition accepted included
  partner-hook:
    protocol: webhook
    url: https://hooks.partner.example/invoices
    secrets: ["file:/etc/flexway-secrets/hooks/current"]
auth:
  issuers:
    - name: customers
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: api
      tenant_claim: org
```

```yaml
# examples/async-amqp1-servicebus/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Invoices, version: 1.0.0 }
servers:
  servicebus: { host: acme-billing.servicebus.windows.net, protocol: amqp, protocolVersion: "1.0" }
channels:
  invoices:
    address: "invoices-{tenant}"                # the tenant's Service Bus topic, created by the operator
    parameters:
      tenant: { x-flexway-bind: tenant }
    bindings: { amqp1: {} }
    messages:
      issued:
        payload: { type: object, required: [invoiceId], properties: { invoiceId: { type: string } } }
  invoicesForPartner:
    address: "invoices-{tenant}/Subscriptions/flexway-partner"   # the subscription, created by the operator
    parameters:
      tenant: { x-flexway-bind: tenant }
    bindings: { amqp1: {} }
    messages:
      issued:
        payload: { type: object, required: [invoiceId], properties: { invoiceId: { type: string } } }
operations:
  publishInvoice: { action: send, channel: { $ref: "#/channels/invoices" }, x-flexway-upstream: billing-bus }
  deliverInvoices:
    action: receive
    channel: { $ref: "#/channels/invoicesForPartner" }
    x-flexway-upstream: billing-bus
    x-flexway-deliver:
      - { upstream: partner-hook, tenant: acme, dead_letter: invoices-dead }   # a dead-letter queue of its own, with problemcode
```

```yaml
# examples/async-amqp1-servicebus/openapi.yaml
openapi: 3.1.0
info: { title: Invoices, version: "1.0.0" }
components:
  securitySchemes:
    customers: { type: openIdConnect, openIdConnectUrl: https://login.example.net/.well-known/openid-configuration }
paths:
  /invoices:
    post:
      operationId: issueInvoice
      x-flexway-publish: { operation: publishInvoice, idempotency: required }   # Idempotency-Key -> the message-id
      security: [ { customers: [invoices:write] } ]
      requestBody:
        required: true
        content:
          application/json: { schema: { type: object, required: [invoiceId], properties: { invoiceId: { type: string } } } }
      responses: { "202": { description: Accepted by Service Bus } }
```

| Ce qui se passe | AMQP 1.0 |
| --- | --- |
| `POST /invoices` | Message réglé par le courtier : `accepted` -> `202` ; `rejected` -> `502 upstream_error` ; hors budget -> `504`, issue inconnue |
| Livraison | Récepteurs concurrents sur l'abonnement (une réplique par message), `accepted` après le `2xx` ; réessais épuisés : l'évènement dans `invoices-dead` avec `problemcode`, jamais le rebut propre au courtier |
| SSE ou WebSocket sur Service Bus | **Refusés au chargement** : aucune source non concurrente sans moyen propre au courtier |
| Jeton Entra ID (CBS) | Hors périmètre de `1.6.0` : clé SAS seulement |

## AMQP 1.0 : consommer RabbitMQ 4 en SSE et WebSocket

> Disponible depuis `1.6.0` ; exemple validé par le binaire et épinglé par les tests.

RabbitMQ 4 parle AMQP 1.0 nativement. Chaque joueur lit **ses** notifications (canal lié au sujet vérifié),
en SSE ou en WebSocket accusé ; tout le monde lit le tableau des scores, public. `amqp1.broker: rabbitmq` dit
comment ouvrir une source **non concurrente** : une file exclusive liée à l'échange, comme le fait déjà
l'adaptateur AMQP 0-9-1 ; une livraison par webhook lit une file `/queues/...` telle quelle. La passerelle s'authentifie
en SASL PLAIN sur TLS ; `EXTERNAL` prend son certificat client (`tls.cert_file`), `ANONYMOUS` n'est admis qu'avec mTLS ou
dans un maillage déclaré. SSE ou WebSocket sur une file `/queues/...` : **refusés au chargement** (ses consommateurs se la partagent).

<!-- example: examples/async-amqp1-rabbitmq/flexway.yaml -->

```yaml
# AMQP 1.0 toward RabbitMQ 4 (1.6.0): each player reads its own notifications (bound to the
# verified subject) in server-sent events or an acknowledged WebSocket; anyone reads the public
# scoreboard. amqp1.broker: rabbitmq says how to read without competing: a queue of the gateway's own,
# bound to the exchange through RabbitMQ 4's management node. The gateway authenticates as its own
# user (SASL PLAIN over TLS); EXTERNAL takes its client certificate instead.
version: 1
contracts:
  openapi: ./openapi.yaml
  asyncapi: ./asyncapi.yaml
upstreams:
  games-bus:
    protocol: amqp
    protocol_version: "1.0"
    brokers: ["amqps://rabbitmq.internal:5671"]
    tls: { server_name: rabbitmq.internal }      # cert_file and key_file with sasl EXTERNAL: the certificate is the identity
    sasl: { mechanism: PLAIN, username: flexway, password: "file:/etc/flexway-secrets/rabbitmq/password" }
    amqp1: { broker: rabbitmq }                  # an exclusive queue bound through the management node
auth:
  issuers:
    - name: players
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: games
      tenant_claim: org
```

```yaml
# examples/async-amqp1-rabbitmq/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Game events, version: 1.0.0 }
channels:
  notifications:
    address: "/exchanges/notifications/players.{sub}"   # a RabbitMQ 4 v2 address: exchange and routing key
    parameters:
      sub: { x-flexway-bind: subject }         # a player reads its own channel, and only it
    bindings: { amqp1: {} }
    messages:
      note:
        payload: { type: object, required: [text], properties: { text: { type: string } } }
  scores:
    address: "/exchanges/scores/public"
    bindings: { amqp1: {} }
    messages:
      # Written as application/cloudevents+json by an outbox: read in structured mode.
      score:
        payload: { type: object, required: [match], properties: { match: { type: string } } }
operations:
  receiveNotifications: { action: receive, channel: { $ref: "#/channels/notifications" }, x-flexway-upstream: games-bus }
  receiveScores:
    action: receive
    channel: { $ref: "#/channels/scores" }
    x-flexway-upstream: games-bus
    x-flexway-public: true                       # read by anyone, anonymous included
```

```yaml
# examples/async-amqp1-rabbitmq/openapi.yaml
openapi: 3.1.0
info: { title: Game events, version: "1.0.0" }
components:
  securitySchemes:
    players: { type: openIdConnect, openIdConnectUrl: https://login.example.net/.well-known/openid-configuration }
paths:
  /me/notifications:
    get:
      operationId: streamMyNotifications
      # WebSocket: settled accepted after {"ack": "<id>"}, released past ack_timeout - delivered again
      x-flexway-subscribe: { operation: receiveNotifications, transport: [sse, websocket], max_streams: 3, ack_timeout: 20s }
      security: [ { players: [] } ]
      responses: { "200": { description: The caller's own notifications } }
  /scores:
    get:
      operationId: streamScores
      x-flexway-subscribe: { operation: receiveScores, transport: [sse], max_anonymous_streams: 500, history: 128 }
      security: []
      responses: { "200": { description: The public scoreboard } }
```

Avec Artemis (`broker: artemis`, attache multicast) ou Solace (`broker: solace`, `topic://`), la même
déclaration ; avec `generic` ou `servicebus`, SSE et WebSocket sont **refusés au chargement**.