Publier et consommer sur un courtier

Guide d'exploitation : configurer, déployer et observer la passerelle. Ce que fait la capacité côté produit : guide produit ; l'appeler : guide développeur.

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.

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

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

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

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

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, 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	RabbitMQ	MQTT 5	JetStream (JetStream ; Core refusé)	Service Bus
Publier via un outil MCP	Kafka	comme Kafka	comme Kafka	JetStream	comme JetStream
SSE en éventail, reprise Last-Event-ID	Kafka	RabbitMQ	MQTT 5 (sans reprise)	NATS	RabbitMQ 4
WebSocket avec accusé	Kafka	RabbitMQ	comme Kafka	NATS (JetStream seul)	RabbitMQ 4 (RabbitMQ 4, Artemis, Solace)
Webhook et file de rebut	Kafka	comme Kafka	MQTT 5 (MQTT 5 seul)	JetStream webhook (JetStream seul)	Service Bus
Lié au tenant	Kafka	comme Kafka	MQTT 5 (attribut tenant)	JetStream, NATS	Service Bus
Lié au sujet	comme RabbitMQ	RabbitMQ	comme RabbitMQ	comme RabbitMQ	RabbitMQ 4
Lié à une claim	comme MQTT 5	comme MQTT 5	MQTT 5	NATS	comme NATS
Canal public	comme RabbitMQ	RabbitMQ	comme RabbitMQ	NATS	RabbitMQ 4
CloudEvents structuré	comme RabbitMQ	RabbitMQ	comme RabbitMQ (toujours en 3.1.1)	NATS	RabbitMQ 4
Identité de la passerelle	Kafka : SCRAM sur TLS, mTLS	RabbitMQ : PLAIN, hôte virtuel	MQTT 5 : PLAIN sur mqtts://	JetStream identifiants, NATS jeton et mot de passe, JetStream webhook nkey	Service Bus : clé SAS, RabbitMQ 4 : PLAIN, EXTERNAL avec un certificat