# Publier et consommer sur un courtier

> Objectif : qu'un service HTTP publie sur un courtier Kafka, AMQP 0-9-1, MQTT, NATS ou AMQP 1.0 sans embarquer de client, et qu'un navigateur, un mobile ou un…

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

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

**Objectif** : qu'un service HTTP publie sur un courtier Kafka, AMQP 0-9-1, MQTT, NATS ou AMQP 1.0 **sans embarquer
de client**, et qu'un navigateur, un mobile ou un partenaire reçoive les évènements **de son tenant** -
derrière les mêmes contrôles qu'une API : identité, portées, débit, quota, trace, audit.

Trois fichiers, chacun son rôle : l'**AsyncAPI** dit les canaux et les messages, l'**OpenAPI** dit
les portes HTTP, `flexway.yaml` dit où sont les courtiers.

```yaml
# flexway.yaml - ne liste aucun canal : il référence le contrat
contracts:
  openapi: ./api/openapi.yaml
  asyncapi: ./api/asyncapi.yaml        # AsyncAPI 3.0.0 ou 3.1.0
upstreams:
  events:
    protocol: kafka                    # ou amqp (amqps://...), mqtt (mqtts://..., protocol_version: 5 ou 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                        # budget d'une publication, accusé compris
    max_in_flight: 512                 # tampon borné : au-delà, 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 }     # lié à l'identité, jamais lu de la requête
    messages:
      created: { payload: { $ref: "./schemas/order-created.json" } }
operations:
  publishOrderCreated:
    action: send
    channel: { $ref: "#/channels/orderCreated" }
    x-flexway-upstream: events
    x-flexway-key: /orderId                  # l'ordre est garanti par clé, et par clé seulement
  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 - les portes, comme toute opération
paths:
  /orders/events:
    post:
      operationId: emitOrderCreated
      x-flexway-publish: { operation: publishOrderCreated, idempotency: required }
      x-flexway-tool: { name: emit_order_created }      # facultatif : aussi un outil MCP
      requestBody:
        content:                       # en ligne : le contrat OpenAPI ne suit pas un $ref vers un fichier
          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 } }
```

**Ce qui se passe** :

| Étape | Ce que fait la passerelle |
| --- | --- |
| Publier (`POST /orders/events`) | Admission complète, corps vérifié contre le schéma du message, évènement **CloudEvents 1.0** (`id` = `Idempotency-Key`, `tenant` de l'identité, `traceparent`), **`202` après l'accusé** du courtier, avec `{"id": ...}` |
| S'abonner en SSE | `GET` avec `Accept: text/event-stream` : un évènement par message de **son** tenant. Une réplique lit chaque canal **une fois**, quel que soit le nombre d'abonnés (`flexway_async_consumers`), et remet à chacun le sien ; un abonné trop lent pour son `buffer` est fermé (`overloaded`, `flexway_async_disconnects_total{reason="slow"}`). Un client qui revient avec `Last-Event-ID` reprend depuis l'historique du canal (`history`, 64 messages) ; au-delà, `412 precondition_failed` : rouvrir sans `Last-Event-ID` |
| S'abonner en WebSocket | Chaque message est un CloudEvent JSON ; le client accuse `{"ack": "<id>"}` ; sans accusé dans `ack_timeout`, fermeture `1008` et nouvelle livraison au retour - **au moins une fois** |
| Livrer en webhook | Une réplique par message, signé Standard Webhooks (`webhook-id` = `id`), réessais bornés, puis le **rebut** avec `problemcode` et `problemstatus` |

**Ce qu'il faut savoir** :

- **`504` = issue inconnue** : le courtier n'a pas accusé dans le budget, le message a pu être écrit.
  Renvoyer **la même** `Idempotency-Key` : les consommateurs dédupliquent sur `source` + `id`.
  `idempotency: required` refuse une publication sans elle (`428`).
- **Ordre** : par clé seulement (`x-flexway-key`). Un webhook lent retient sa partition, jamais les
  autres livraisons - borné par son coupe-circuit et ses réessais.
- **Le tenant** : un canal lié au tenant (`x-flexway-bind`) ou un message qui déclare l'en-tête
  `tenant` - sans l'un ni l'autre, la consommation est **refusée au démarrage**. Un en-tête `ce-*`
  envoyé par l'appelant est refusé (`400`) : les attributs sont ceux de la passerelle.
- **Un canal par personne** : `x-flexway-bind: subject` (ou `claim:<nom>`, une claim du jeton
  vérifié) - `players.{sub}.notifications` n'est lu que par ce sujet ; une valeur hors des lettres,
  chiffres, `_` et `-` est refusée (`403`).
- **Un flux public** (anonymes, écran d'affichage) : `x-flexway-public: true` sur l'opération
  `receive`, une porte `security: []`, `max_anonymous_streams` pour borner tous les anonymes
  ensemble ; refusé si le canal lie ou filtre quoi que ce soit.
- **Mode structuré** : un producteur hors de la passerelle (une outbox) peut écrire en
  `application/cloudevents+json` ; la passerelle lit les deux modes, selon le type de contenu.
- **ACL minimales du courtier**, pour le principal de la passerelle : écrire sur les sujets des
  opérations `send` (et le rebut), lire ceux des `receive`, gérer ses groupes `flexway.ws.*` et
  `flexway.deliver.*` (Kafka) ou déclarer ses files (AMQP). **Jamais** créer un sujet : la passerelle
  n'en crée pas - un tenant sans sujet répond `502 upstream_error`.
- **Un courtier injoignable** ne bloque jamais le démarrage : son coupe-circuit s'ouvre (`503`), les
  autres opérations servent. Un client inchangé est gardé au rechargement.

**Vérifier** :

```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` (administration) montre `async` : canaux, sens, isolation (`bind`, `public`), portes
(publication ou abonnement, transports, plafonds) et livraisons, jamais une adresse résolue pour un
tenant ; `flexway_async_publish_total` et `flexway_async_deliver_total` comptent par opération,
tenant et issue ; une publication en attente d'accusé compte dans `flexway_requests_in_flight` du
courtier.

**Charges binaires** (`1.6.0`) : un message peut être **Avro** ou **Protobuf** (`payload.schemaFormat`)
ou **CBOR** (`contentType: application/cbor` et un schéma JSON). Le schéma est compilé au chargement ; la
publication exige le `Content-Type` déclaré (`415` sinon, `422` si la porte déclare un autre corps) et
décode le corps sous son schéma si l'opération AsyncAPI le demande (`x-flexway-validation: full`, choix
de l'opération : un décodage par message) ; un abonné SSE ou un webhook reçoit le message **en JSON**
(`datacontenttype: application/json`), une WebSocket le reçoit tel quel si sa porte déclare
`frames: binary`. Le format filaire de Confluent (octet magique, identifiant du schéma) se lit dans un
registre déclaré (`schema_registry`), le schéma devant être enregistré sous le sujet du tenant -
exemple complet en [Charges Avro sous schéma, registre de schémas](https://flexway.io/docs/operations-examples-brokers/#charges-avro-sous-sch%C3%A9ma-registre-de-sch%C3%A9mas). Un **producteur Kafka
ordinaire** (un sérialiseur de Confluent, sans en-tête CloudEvents) est lu sur un canal **lié** au
tenant ou **public** dont le seul message déclare ce format : la passerelle construit l'enveloppe
(`id` = `<sujet>/<partition>/<position>`, horodatage Kafka), le tenant venant du canal lié seul.

**Clients Kafka natifs** (`1.6.0`) : au-delà d'un appelant HTTP, un client Kafka
de n'importe quelle bibliothèque se branche sur l'**écouteur Kafka** (`listeners.kafka`) ; une
opération marquée `x-flexway-kafka-proxy` lui sert le canal comme sujet d'un cluster virtuel, aux noms
écrits dans l'espace de son tenant, sous les identités de HTTP (jeton en OAUTHBEARER, clé d'API en PLAIN,
certificat client) - exemple complet en [Clients Kafka natifs : clusters virtuels par tenant](https://flexway.io/docs/operations-kafka-clients/).

binaires et registre, écouteur Kafka sur `kmsg`, NATS Core et JetStream (nats.go) et AMQP 1.0 (go-amqp).*

**Chaque courtier, chaque cas** - un exemple complet et validé par courtier en [Exemples : Kafka, RabbitMQ, MQTT, Avro, NATS, AMQP 1.0](https://flexway.io/docs/operations-examples-brokers/), NATS et
AMQP 1.0 (`1.6.0`) compris, des fichiers de `examples/` validés comme les autres. « Comme » renvoie à la même déclaration, qui ne dépend pas du courtier.

| Cas | Kafka | AMQP 0-9-1 | MQTT | NATS (`1.6.0`) | AMQP 1.0 (`1.6.0`) |
| --- | --- | --- | --- | --- | --- |
| Publier depuis HTTP | [Kafka](https://flexway.io/docs/operations-examples-brokers/#kafka--publier-diffuser-livrer) | [RabbitMQ](https://flexway.io/docs/operations-examples-brokers/#rabbitmq-en-amqp-0-9-1--canal-par-personne-flux-public) | [MQTT 5](https://flexway.io/docs/operations-examples-brokers/#mqtt-5--claim-attribut-tenant-webhook) | [JetStream](https://flexway.io/docs/operations-examples-brokers/#nats-jetstream--publier-depuis-http-ou-un-outil-mcp) (JetStream ; Core refusé) | [Service Bus](https://flexway.io/docs/operations-examples-brokers/#amqp-10--publier-vers-azure-service-bus-livrer-depuis-un-abonnement) |
| Publier via un outil MCP | [Kafka](https://flexway.io/docs/operations-examples-brokers/#kafka--publier-diffuser-livrer) | comme [Kafka](https://flexway.io/docs/operations-examples-brokers/#kafka--publier-diffuser-livrer) | comme [Kafka](https://flexway.io/docs/operations-examples-brokers/#kafka--publier-diffuser-livrer) | [JetStream](https://flexway.io/docs/operations-examples-brokers/#nats-jetstream--publier-depuis-http-ou-un-outil-mcp) | comme [JetStream](https://flexway.io/docs/operations-examples-brokers/#nats-jetstream--publier-depuis-http-ou-un-outil-mcp) |
| SSE en éventail, reprise `Last-Event-ID` | [Kafka](https://flexway.io/docs/operations-examples-brokers/#kafka--publier-diffuser-livrer) | [RabbitMQ](https://flexway.io/docs/operations-examples-brokers/#rabbitmq-en-amqp-0-9-1--canal-par-personne-flux-public) | [MQTT 5](https://flexway.io/docs/operations-examples-brokers/#mqtt-5--claim-attribut-tenant-webhook) (sans reprise) | [NATS](https://flexway.io/docs/operations-examples-brokers/#nats--sse-et-websocket-li%C3%A9s-au-tenant-jokers-nats-core) | [RabbitMQ 4](https://flexway.io/docs/operations-examples-brokers/#amqp-10--consommer-rabbitmq-4-en-sse-et-websocket) |
| WebSocket avec accusé | [Kafka](https://flexway.io/docs/operations-examples-brokers/#kafka--publier-diffuser-livrer) | [RabbitMQ](https://flexway.io/docs/operations-examples-brokers/#rabbitmq-en-amqp-0-9-1--canal-par-personne-flux-public) | comme [Kafka](https://flexway.io/docs/operations-examples-brokers/#kafka--publier-diffuser-livrer) | [NATS](https://flexway.io/docs/operations-examples-brokers/#nats--sse-et-websocket-li%C3%A9s-au-tenant-jokers-nats-core) (JetStream seul) | [RabbitMQ 4](https://flexway.io/docs/operations-examples-brokers/#amqp-10--consommer-rabbitmq-4-en-sse-et-websocket) (RabbitMQ 4, Artemis, Solace) |
| Webhook et file de rebut | [Kafka](https://flexway.io/docs/operations-examples-brokers/#kafka--publier-diffuser-livrer) | comme [Kafka](https://flexway.io/docs/operations-examples-brokers/#kafka--publier-diffuser-livrer) | [MQTT 5](https://flexway.io/docs/operations-examples-brokers/#mqtt-5--claim-attribut-tenant-webhook) (MQTT 5 seul) | [JetStream webhook](https://flexway.io/docs/operations-examples-brokers/#nats-jetstream--webhook-depuis-un-consommateur-durable) (JetStream seul) | [Service Bus](https://flexway.io/docs/operations-examples-brokers/#amqp-10--publier-vers-azure-service-bus-livrer-depuis-un-abonnement) |
| Lié au tenant | [Kafka](https://flexway.io/docs/operations-examples-brokers/#kafka--publier-diffuser-livrer) | comme [Kafka](https://flexway.io/docs/operations-examples-brokers/#kafka--publier-diffuser-livrer) | [MQTT 5](https://flexway.io/docs/operations-examples-brokers/#mqtt-5--claim-attribut-tenant-webhook) (attribut `tenant`) | [JetStream](https://flexway.io/docs/operations-examples-brokers/#nats-jetstream--publier-depuis-http-ou-un-outil-mcp), [NATS](https://flexway.io/docs/operations-examples-brokers/#nats--sse-et-websocket-li%C3%A9s-au-tenant-jokers-nats-core) | [Service Bus](https://flexway.io/docs/operations-examples-brokers/#amqp-10--publier-vers-azure-service-bus-livrer-depuis-un-abonnement) |
| Lié au sujet | comme [RabbitMQ](https://flexway.io/docs/operations-examples-brokers/#rabbitmq-en-amqp-0-9-1--canal-par-personne-flux-public) | [RabbitMQ](https://flexway.io/docs/operations-examples-brokers/#rabbitmq-en-amqp-0-9-1--canal-par-personne-flux-public) | comme [RabbitMQ](https://flexway.io/docs/operations-examples-brokers/#rabbitmq-en-amqp-0-9-1--canal-par-personne-flux-public) | comme [RabbitMQ](https://flexway.io/docs/operations-examples-brokers/#rabbitmq-en-amqp-0-9-1--canal-par-personne-flux-public) | [RabbitMQ 4](https://flexway.io/docs/operations-examples-brokers/#amqp-10--consommer-rabbitmq-4-en-sse-et-websocket) |
| Lié à une claim | comme [MQTT 5](https://flexway.io/docs/operations-examples-brokers/#mqtt-5--claim-attribut-tenant-webhook) | comme [MQTT 5](https://flexway.io/docs/operations-examples-brokers/#mqtt-5--claim-attribut-tenant-webhook) | [MQTT 5](https://flexway.io/docs/operations-examples-brokers/#mqtt-5--claim-attribut-tenant-webhook) | [NATS](https://flexway.io/docs/operations-examples-brokers/#nats--sse-et-websocket-li%C3%A9s-au-tenant-jokers-nats-core) | comme [NATS](https://flexway.io/docs/operations-examples-brokers/#nats--sse-et-websocket-li%C3%A9s-au-tenant-jokers-nats-core) |
| Canal public | comme [RabbitMQ](https://flexway.io/docs/operations-examples-brokers/#rabbitmq-en-amqp-0-9-1--canal-par-personne-flux-public) | [RabbitMQ](https://flexway.io/docs/operations-examples-brokers/#rabbitmq-en-amqp-0-9-1--canal-par-personne-flux-public) | comme [RabbitMQ](https://flexway.io/docs/operations-examples-brokers/#rabbitmq-en-amqp-0-9-1--canal-par-personne-flux-public) | [NATS](https://flexway.io/docs/operations-examples-brokers/#nats--sse-et-websocket-li%C3%A9s-au-tenant-jokers-nats-core) | [RabbitMQ 4](https://flexway.io/docs/operations-examples-brokers/#amqp-10--consommer-rabbitmq-4-en-sse-et-websocket) |
| CloudEvents structuré | comme [RabbitMQ](https://flexway.io/docs/operations-examples-brokers/#rabbitmq-en-amqp-0-9-1--canal-par-personne-flux-public) | [RabbitMQ](https://flexway.io/docs/operations-examples-brokers/#rabbitmq-en-amqp-0-9-1--canal-par-personne-flux-public) | comme [RabbitMQ](https://flexway.io/docs/operations-examples-brokers/#rabbitmq-en-amqp-0-9-1--canal-par-personne-flux-public) (toujours en 3.1.1) | [NATS](https://flexway.io/docs/operations-examples-brokers/#nats--sse-et-websocket-li%C3%A9s-au-tenant-jokers-nats-core) | [RabbitMQ 4](https://flexway.io/docs/operations-examples-brokers/#amqp-10--consommer-rabbitmq-4-en-sse-et-websocket) |
| Identité de la passerelle | [Kafka](https://flexway.io/docs/operations-examples-brokers/#kafka--publier-diffuser-livrer) : SCRAM sur TLS, mTLS | [RabbitMQ](https://flexway.io/docs/operations-examples-brokers/#rabbitmq-en-amqp-0-9-1--canal-par-personne-flux-public) : PLAIN, hôte virtuel | [MQTT 5](https://flexway.io/docs/operations-examples-brokers/#mqtt-5--claim-attribut-tenant-webhook) : PLAIN sur `mqtts://` | [JetStream](https://flexway.io/docs/operations-examples-brokers/#nats-jetstream--publier-depuis-http-ou-un-outil-mcp) identifiants, [NATS](https://flexway.io/docs/operations-examples-brokers/#nats--sse-et-websocket-li%C3%A9s-au-tenant-jokers-nats-core) jeton et mot de passe, [JetStream webhook](https://flexway.io/docs/operations-examples-brokers/#nats-jetstream--webhook-depuis-un-consommateur-durable) nkey | [Service Bus](https://flexway.io/docs/operations-examples-brokers/#amqp-10--publier-vers-azure-service-bus-livrer-depuis-un-abonnement) : clé SAS, [RabbitMQ 4](https://flexway.io/docs/operations-examples-brokers/#amqp-10--consommer-rabbitmq-4-en-sse-et-websocket) : PLAIN, EXTERNAL avec un certificat |