Example: MQTT 5
An MQTT 5 broker: a channel bound to a claim, the tenant attribute, webhook delivery through a shared subscription.
ExampleOperationsSince version 1.5.0
Operations guide: configure, deploy and observe the gateway. What the feature does, product side: product guide; calling it: developer guide.
MQTT 5: a claim, the tenant attribute, a webhook
An operator reads the telemetry of its own site - the channel is bound to the token's site claim -,
without resumption (history: 0); every tenant's alerts go through one shared topic, each carrying the
tenant attribute: consumption is filtered on it, and acme's alerts leave as webhooks through an MQTT 5
shared subscription (one replica per 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) } }| Case | What the gateway does |
|---|---|
POST /telemetry |
202 after the QoS 1 PUBACK; Idempotency-Key required |
GET /telemetry |
The token's sites/<site>/telemetry channel only; without a site claim, refused |
POST /alerts |
The tenant attribute comes from the identity, never from the body |
Delivery to ops-hook |
A shared subscription, filtered on tenant = acme, dead letters in plant/alerts-dead; refused at load on MQTT 3.1.1 |
| mTLS instead of the password | tls.cert_file and tls.key_file mounted, no sasl |
Run
Save the
flexway.yamlshown and the contracts it references in one directory, next to it (examples/async-mqtt/in the release sources).Validate offline, without a broker or a network - the validator ships in the published image:
docker run --rm -v "$PWD:/etc/flexway:ro" ghcr.io/flexwiz/flexway:<version> validate -vMount the secrets the file references (
file:/etc/flexway-secrets/...), then start: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>
Verify
validate -vexits0and lists every resolved operation; a refusal names the faulty field and exits1.- Once started,
curl -s localhost:9090/readyzanswers200{"status":"ready","operations":<n>}; a missing secret makes the start refuse, naming it. - Each row of the case table replays from the developer side: Publish and subscribe, Receive signed webhooks.