Exemple : MQTT 5

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.

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

# 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

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

# 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

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