Exemples : NATS et JetStream

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.

NATS JetStream : publier depuis HTTP ou un outil MCP

Un service HTTP ou un agent publie dans un flux JetStream sans client NATS ; 202 seulement après le PubAck du flux, l'Idempotency-Key devenue Nats-Msg-Id - le flux dédoublonne un renvoi après un 504. La passerelle ne crée aucun flux : le sujet doit être capturé par un flux existant, sinon 502 upstream_error.

# NATS JetStream behind the gateway's controls (1.6.0): an HTTP door and an MCP tool publish
# to a stream without a NATS client - 202 only after the stream's PubAck, the Idempotency-Key as
# Nats-Msg-Id. The gateway creates no stream: a subject no stream captures answers 502.
version: 1
contracts:
  openapi: ./openapi.yaml
  asyncapi: ./asyncapi.yaml
upstreams:
  orders-nats:
    protocol: nats
    brokers: ["tls://nats-0.nats:4222", "tls://nats-1.nats:4222"]   # nats:// or tls://, never a credential in the URL
    tls: { server_name: nats.internal }
    nats:
      credentials: "file:/etc/flexway-secrets/nats/flexway.creds"   # the gateway's NATS user: JWT and nkey seed
      jetstream: true                            # 202 after the stream's PubAck; false (NATS Core) refuses a publication
    timeout: 2s                                  # one publication's budget, the PubAck included
    max_in_flight: 512                           # bounded buffer: past it, 503 overloaded
auth:
  issuers:
    - name: customers
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: api
      tenant_claim: org                          # the tenant every subject below is bound to
agents:
  mcp:
    resources:
      api.example.com: https://api.example.com/mcp

# examples/async-nats-jetstream/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Orders, version: 1.0.0 }
channels:
  orderCreated:
    address: "orders.{tenant}.created"          # captured by the ORDERS stream (orders.>): the gateway creates no stream
    parameters:
      tenant: { x-flexway-bind: tenant }
    messages:
      created:
        contentType: application/json
        payload: { type: object, required: [orderId], properties: { orderId: { type: string } } }
operations:
  publishOrderCreated:
    action: send
    channel: { $ref: "#/channels/orderCreated" }
    x-flexway-upstream: orders-nats

# examples/async-nats-jetstream/openapi.yaml
openapi: 3.1.0
info: { title: Orders, version: "1.0.0" }
components:
  securitySchemes:
    customers: { type: openIdConnect, openIdConnectUrl: https://login.example.net/.well-known/openid-configuration }
paths:
  /orders/events:
    post:
      operationId: emitOrderCreated
      x-flexway-publish: { operation: publishOrderCreated, idempotency: required }   # Idempotency-Key -> Nats-Msg-Id
      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 } } } }
      responses: { "202": { description: Acknowledged by the stream } }

Ce qui se passe	NATS JetStream
POST /orders/events	Admission, corps vérifié, CloudEvents en mode binaire (en-têtes ce-*), publication sur orders.acme.created, 202 après le PubAck
Le même appel en outil MCP (emit_order_created)	Même décision, même accusé
jetstream: false (NATS Core)	> [!WARNING]

Refusé au chargement : Core n'accuse rien, un 202 mentirait | | Aucun flux ne capture le sujet | 502 upstream_error (« no responders »), jamais un 202 |

NATS : SSE et WebSocket liés au tenant, jokers, NATS Core

Un tableau de bord lit tous les évènements de commande de son tenant d'un seul flux, grâce à un joker inclus dans le canal après le segment lié ; un opérateur lit les mesures de son site sur un NATS Core (SSE seul, au plus une fois) ; un écran public lit l'état sans jeton. Chaque message reçu est comparé au motif du canal, segment par segment : un courtier qui en enverrait trop ne fait jamais fuir un tenant.

# NATS in server-sent events and WebSocket (1.6.0): a dashboard reads every order event of
# its tenant through a wildcard written in the channel after the bound segment; an operator reads its
# own site's readings from NATS Core (server-sent events only, at most once); a public status is read
# without a token. Two brokers, two identities.
version: 1
contracts:
  openapi: ./openapi.yaml
  asyncapi: ./asyncapi.yaml
upstreams:
  orders-nats:
    protocol: nats
    brokers: ["tls://nats-0.nats:4222"]
    tls: { server_name: nats.internal }
    nats:
      token: "file:/etc/flexway-secrets/nats/token"   # a token, by reference - over TLS only
      jetstream: true
  metrics-core:
    protocol: nats
    brokers: ["nats://nats-edge.plant:4222"]
    sasl: { mechanism: PLAIN, username: flexway, password: "file:/etc/flexway-secrets/nats-edge/password" }
    tls: { server_name: nats-edge.plant }
    nats:
      jetstream: false                           # NATS Core: server-sent events only - no publication, WebSocket nor webhook
auth:
  issuers:
    - name: operators
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: api
      tenant_claim: org

# examples/async-nats-streams/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Orders and metrics, version: 1.0.0 }
channels:
  orderFeed:
    address: "orders.{tenant}.>"                # the wildcard IN the channel, after the segment bound to the tenant
    parameters:
      tenant: { x-flexway-bind: tenant }
    messages:
      any:
        payload: { type: object, required: [orderId], properties: { orderId: { type: string } } }
  siteMetrics:
    address: "sites.{site}.metrics.*"           # bound to a claim of the token: an operator reads its own site
    parameters:
      site: { x-flexway-bind: "claim:site" }
    messages:
      # Written as application/cloudevents+json by a producer outside the gateway: read in structured mode.
      reading:
        payload: { type: object, required: [value], properties: { value: { type: number } } }
  status:
    address: status.public                       # binds nothing, filters nothing: declared public below
    messages:
      state:
        payload: { type: object, required: [state], properties: { state: { type: string } } }
operations:
  receiveOrders: { action: receive, channel: { $ref: "#/channels/orderFeed" }, x-flexway-upstream: orders-nats }
  receiveMetrics: { action: receive, channel: { $ref: "#/channels/siteMetrics" }, x-flexway-upstream: metrics-core }
  receiveStatus:
    action: receive
    channel: { $ref: "#/channels/status" }
    x-flexway-upstream: orders-nats
    x-flexway-public: true                       # read by anyone, anonymous included

# examples/async-nats-streams/openapi.yaml
openapi: 3.1.0
info: { title: Orders and metrics, version: "1.0.0" }
components:
  securitySchemes:
    operators: { type: openIdConnect, openIdConnectUrl: https://login.example.net/.well-known/openid-configuration }
paths:
  /orders/stream:
    get:
      operationId: streamOrders
      # SSE: one ordered consumer per channel and replica; WebSocket: one durable consumer per subscriber, AckWait = ack_timeout
      x-flexway-subscribe: { operation: receiveOrders, transport: [sse, websocket], max_streams: 2, ack_timeout: 30s, history: 64 }
      security: [ { operators: [orders:read] } ]
      responses: { "200": { description: The caller's tenant's orders } }
  /metrics/stream:
    get:
      operationId: streamMetrics
      x-flexway-subscribe: { operation: receiveMetrics, transport: [sse], history: 0 }   # NATS Core: SSE only
      security: [ { operators: [metrics:read] } ]
      responses: { "200": { description: The readings of the caller's site } }
  /status:
    get:
      operationId: streamStatus
      x-flexway-subscribe: { operation: receiveStatus, transport: [sse], max_anonymous_streams: 1000 }
      security: []
      responses: { "200": { description: The public status } }

Règle	Comportement
Joker	* et > seulement dans un canal lu, après les segments liés, > en dernier - orders.>.{tenant} est refusé ; une valeur liée ne contient jamais ., * ni >. Même règle pour + et # en MQTT, et pour * et # dans la clé de liaison AMQP 0-9-1 d'un échange (audit.{tenant}.#, servi depuis 1.5.0)
SSE	Un consommateur ordonné JetStream (ou un abonnement Core) par canal et par réplique, quel que soit le nombre d'abonnés ; reprise Last-Event-ID dans l'historique du canal, id = séquence du flux
WebSocket	Un consommateur durable par abonné, AckWait = ack_timeout ; refusé au chargement sur NATS Core
CloudEvents	Binaire (ce-*) ou structuré selon le type de contenu ; l'attribut tenant filtré dans les en-têtes, jamais dans data

NATS JetStream : webhook depuis un consommateur durable

Un partenaire reçoit les factures de son tenant en webhook signé, sans accès à NATS : un consommateur durable partagé par les répliques (une réplique par message), accusé après le 2xx, rebut déclaré au bout des réessais. Aucune porte HTTP : le contrat OpenAPI n'est pas nécessaire.

# NATS JetStream delivered by webhook (1.6.0): a partner receives its tenant's invoices as
# signed webhooks, without access to NATS - a durable consumer the replicas share (one replica per
# message), acknowledged after the 2xx, the declared dead letter once the retries are spent. No HTTP
# door: no OpenAPI contract. Refused on NATS Core, which redelivers nothing.
version: 1
contracts:
  asyncapi: ./asyncapi.yaml
upstreams:
  billing-nats:
    protocol: nats
    brokers: ["tls://nats-0.nats:4222"]
    tls: { server_name: nats.internal }
    nats:
      nkey_seed: "file:/etc/flexway-secrets/nats/flexway.nk"   # an nkey seed alone, by reference
      jetstream: true                            # a delivery needs JetStream: a durable consumer the replicas share
  partner-hook:
    protocol: webhook
    url: https://hooks.partner.example/invoices
    secrets: ["file:/etc/flexway-secrets/hooks/current"]
    retries: { max: 8, max_delay: 10m }

# examples/async-nats-webhook/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Invoices, version: 1.0.0 }
channels:
  invoiceIssued:
    address: "invoices.{tenant}.issued"
    parameters:
      tenant: { x-flexway-bind: tenant }       # a delivery has no caller: bound to the tenant only
    messages:
      issued:
        payload: { type: object, required: [invoiceId], properties: { invoiceId: { type: string } } }
operations:
  receiveInvoices:
    action: receive
    channel: { $ref: "#/channels/invoiceIssued" }
    x-flexway-upstream: billing-nats
    x-flexway-deliver:
      - { upstream: partner-hook, tenant: acme, dead_letter: invoices.dead }   # a subject a stream captures

Sur NATS Core, la même livraison est refusée au chargement : rien ne relivre un message que la cible n'a pas pris.

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-nats-jetstream/, examples/async-nats-streams/, examples/async-nats-webhook/ 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.