Exemples : AMQP 1.0 (Azure Service Bus, RabbitMQ 4)

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.

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

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.

# 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

# 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

# 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

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

# 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

# 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

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

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-amqp1-servicebus/, examples/async-amqp1-rabbitmq/ 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.