# Exemple : MQTT 5 > Un courtier MQTT 5 : canal lié à une claim, attribut tenant, livraison en webhook par abonnement partagé. Exemple - Depuis la version 1.5.0 - Flexway v1.7.1 - https://flexway.io/docs/operations-examples-mqtt/ > Guide d'exploitation : configurer, déployer et observer la passerelle. Ce que fait la fonctionnalité côté produit : [guide produit](https://flexway.io/docs/product-overview/) ; l'appeler : [guide développeur](https://flexway.io/docs/developer-overview/). ## 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). ```yaml # 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 ``` ```yaml # 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 } ``` ```yaml # 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//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 : ```bash docker run --rm -v "$PWD:/etc/flexway:ro" ghcr.io/flexwiz/flexway: validate -v ``` 3. Monter les secrets que le fichier référence (`file:/etc/flexway-secrets/...`), puis démarrer : ```bash 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: ``` ## 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":}` ; 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](https://flexway.io/docs/developer-async/), [Recevoir des webhooks signés](https://flexway.io/docs/developer-webhooks/).