Subscriptions, filters and steps

An event subscription connects two things you already have: an event source and a channel. Its events go from the source into the channel, and when the channel has a type, on to the system the channel delivers to. You write no route, and the subscription is an integration like any other, with versions and a deployment per environment.

Bind a source to a channel#

apiVersion: weik.io/v1
kind: event-subscription
metadata: { name: files-to-invoices, project: finance }
spec:
  source: invoice-files
  channel: invoice-created

$ weikio apply -f ./finance/events/
$ weikio deploy finance/files-to-invoices --env production

Every event the invoice-files source brings in now also lands in invoice-created, where the integrations that read that channel get it. The forwarded event has an ID of its own. It also keeps the original’s CloudEvents type, source, subject and time, so a reader can tell where it came from.

Filter#

filter lets through only the events that match. Every header you list must have the value you give. The header Weikio-Source has the integration that published an event, written <project>/<integration>, and a subscription passes it on unchanged. In a channel that more than one integration feeds, it tells their events apart. A subscription’s source can be an event source or a channel of the project:

apiVersion: weik.io/v1
kind: event-subscription
metadata: { name: alert-on-invoices, project: finance }
spec:
  source: invoice-created
  channel: invoice-alerts
  filter:
    headers: { Weikio-Source: finance/erp-invoice-export }

invoice-created has the events that finance/erp-invoice-export publishes and the file events that files-to-invoices forwards, which keep finance/invoice-files as their source. This subscription passes on only the ERP export’s events.

jsonPath filters by the body: an event passes when the JSON path matches at least one node of it. Headers and the JSON path can be combined, and an event must pass both:

spec:
  source: invoice-files
  channel: invoice-alerts
  filter:
    jsonPath: "$.lines[?(@.amount > 10000)]"

An event that doesn’t pass is left out by this subscription only, and the subscription counts it in a log line. Other subscriptions and readers of the source still get it.

Steps#

steps changes events on the way, with a list of kamelet actions between the source and the channel, as in a Camel K Pipe. Each step is a kamelet from Camel’s catalog or from your project, with its parameters. The filter runs first, then the steps in order:

spec:
  source: invoice-files
  channel: invoice-created
  filter:
    jsonPath: "$.lines[?(@.amount > 10000)]"
  steps:
    - kamelet: json-deserialize-action
      parameters: {}

When a forward fails#

A subscription receives its source’s events through a durable reader. When a forward fails, the event is delivered to the subscription again after the read’s retry delays, and after the last attempt it goes to the environment’s dead letters. errorHandler: log writes the failure once at ERROR and counts the event as handled instead. errorHandler: { sink: { channel: failed-invoices } } sends the failed event to another channel, and { sink: { http: https://example.com/failed } } to a URL.

Fan out to several destinations#

Several subscriptions can read one source, and a subscription can read a channel that another one fills. In the examples above, files-to-invoices sends every file event to invoice-created, and alert-on-invoices passes the ERP export’s events of that channel on to invoice-alerts, a channel with the type slack that posts each one into a Slack channel. Delivery and CloudEvents shows how a channel with a type delivers.

$ weikio get eventsubscriptions --project finance

The list has each subscription with its source, its channel and its state in each environment you can read.