Exemple : Kafka

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

Kafka : publier, diffuser, livrer

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

# 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

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

# 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

Lancer

1. Enregistrer dans un répertoire le flexway.yaml montré et les contrats qu'il référence, à côté de lui (examples/async-kafka/ dans les sources de la release).

2. Valider hors ligne, sans courtier ni réseau - le validateur est dans l'image publiée :

   docker run --rm -v "$PWD:/etc/flexway:ro" ghcr.io/flexwiz/flexway:<version> validate -v

3. Monter les secrets que le fichier référence (file:/etc/flexway-secrets/...), puis démarrer :

   docker run --rm -p 8080:8080 -p 9090:9090 \
     -v "$PWD:/etc/flexway:ro" -v "$PWD/secrets:/etc/flexway-secrets:ro" \
     ghcr.io/flexwiz/flexway:<version>

Vérifier

1. validate -v sort en 0 et liste chaque opération résolue ; un refus nomme le champ fautif et sort en 1.
2. Une fois démarrée, curl -s localhost:9090/readyz répond 200 {"status":"ready","operations":<n>} ; un secret absent fait refuser le démarrage, qui le nomme.
3. Chaque ligne du tableau des cas se rejoue avec le développeur : Publier et s'abonner, Recevoir des webhooks signés.