How-To: Route messages to different event handlers

Learn how to route messages from a topic to different event handlers based on CloudEvent fields

Preview feature

Pub/Sub message routing is currently in preview.

Introduction

Content-based routing is a messaging pattern that utilizes a DSL instead of imperative application code. PubSub routing is an implementation of this pattern that allows developers to use expressions to route CloudEvents based on their contents to different URIs/paths and event handlers in your application. If no route matches, then an optional default route is used. This becomes useful as your applications expands to support multiple event versions, or special cases. Routing can be implemented with code; however, keeping routing rules external from the application can improve portability.

This feature is available to both the declarative and programmatic subscription approaches.

Enable message routing

This is a preview feature. To enable it, add the PubSub.Routing feature entry to your application configuration like so:

apiVersion: dapr.io/v1alpha1
kind: Configuration
metadata:
  name: pubsubroutingconfig
spec:
  features:
    - name: PubSub.Routing
      enabled: true

Learn more about enabling preview features.

Declarative subscription

For declarative subscriptions, you must use dapr.io/v2alpha1 as the apiVersion. Here is an example of subscriptions.yaml using routing.

apiVersion: dapr.io/v2alpha1
kind: Subscription
metadata:
  name: myevent-subscription
spec:
  pubsubname: pubsub
  topic: inventory
  routes:
    rules:
      - match: event.type == "widget"
        path: /widgets
      - match: event.type == "gadget"
        path: /gadgets
    default: /products
scopes:
  - app1
  - app2

Programmatic subscription

Alternatively, the programattic approach varies slightly in that the routes structure is returned instead of route. The JSON structure matches the declarative YAML.

• Python
• Node
• C#
• Go
• PHP

import flask
from flask import request, jsonify
from flask_cors import CORS
import json
import sys
app = flask.Flask(__name__)
CORS(app)
@app.route('/dapr/subscribe', methods=['GET'])
def subscribe():
    subscriptions = [
      {
        'pubsubname': 'pubsub',
        'topic': 'inventory',
        'routes': {
          'rules': [
            {
              'match': 'event.type == "widget"',
              'path': '/widgets'
            },
            {
              'match': 'event.type == "gadget"',
              'path': '/gadgets'
            },
          ],
          'default': '/products'
        }
      }]
    return jsonify(subscriptions)
@app.route('/products', methods=['POST'])
def ds_subscriber():
    print(request.json, flush=True)
    return json.dumps({'success':True}), 200, {'ContentType':'application/json'}
app.run()

const express = require('express')
const bodyParser = require('body-parser')
const app = express()
app.use(bodyParser.json({ type: 'application/*+json' }));
const port = 3000
app.get('/dapr/subscribe', (req, res) => {
  res.json([
    {
      pubsubname: "pubsub",
      topic: "inventory",
      routes: {
        rules: [
          {
            match: 'event.type == "widget"',
            path: '/widgets'
          },
          {
            match: 'event.type == "gadget"',
            path: '/gadgets'
          },
        ],
        default: '/products'
      }
    }
  ]);
})
app.post('/products', (req, res) => {
  console.log(req.body);
  res.sendStatus(200);
});
app.listen(port, () => console.log(`consumer app listening on port ${port}!`))

        [Topic("pubsub", "inventory", "event.type ==\"widget\"", 1)]
        [HttpPost("widgets")]
        public async Task<ActionResult<Stock>> HandleWidget(Widget widget, [FromServices] DaprClient daprClient)
        {
            // Logic
            return stock;
        }
        [Topic("pubsub", "inventory", "event.type ==\"gadget\"", 2)]
        [HttpPost("gadgets")]
        public async Task<ActionResult<Stock>> HandleGadget(Gadget gadget, [FromServices] DaprClient daprClient)
        {
            // Logic
            return stock;
        }
        [Topic("pubsub", "inventory")]
        [HttpPost("products")]
        public async Task<ActionResult<Stock>> HandleProduct(Product product, [FromServices] DaprClient daprClient)
        {
            // Logic
            return stock;
        }

package main
import (
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "github.com/gorilla/mux"
)
const appPort = 3000
type subscription struct {
    PubsubName string            `json:"pubsubname"`
    Topic      string            `json:"topic"`
    Metadata   map[string]string `json:"metadata,omitempty"`
    Routes     routes            `json:"routes"`
}
type routes struct {
    Rules   []rule `json:"rules,omitempty"`
    Default string `json:"default,omitempty"`
}
type rule struct {
    Match string `json:"match"`
    Path  string `json:"path"`
}
// This handles /dapr/subscribe
func configureSubscribeHandler(w http.ResponseWriter, _ *http.Request) {
    t := []subscription{
        {
            PubsubName: "pubsub",
            Topic:      "inventory",
            Routes: routes{
                Rules: []rule{
                    {
                        Match: `event.type == "widget"`,
                        Path:  "/widgets",
                    },
                    {
                        Match: `event.type == "gadget"`,
                        Path:  "/gadgets",
                    },
                },
                Default: "/products",
            },
        },
    }
    w.WriteHeader(http.StatusOK)
    json.NewEncoder(w).Encode(t)
}
func main() {
    router := mux.NewRouter().StrictSlash(true)
    router.HandleFunc("/dapr/subscribe", configureSubscribeHandler).Methods("GET")
    log.Fatal(http.ListenAndServe(fmt.Sprintf(":%d", appPort), router))
}

<?php
require_once __DIR__.'/vendor/autoload.php';
$app = \Dapr\App::create(configure: fn(\DI\ContainerBuilder $builder) => $builder->addDefinitions(['dapr.subscriptions' => [
    new \Dapr\PubSub\Subscription(pubsubname: 'pubsub', topic: 'inventory', routes: (
      rules: => [
        ('match': 'event.type == "widget"', path: '/widgets'),
        ('match': 'event.type == "gadget"', path: '/gadgets'),
      ]
      default: '/products')),
]]));
$app->post('/products', function(
    #[\Dapr\Attributes\FromBody]
    \Dapr\PubSub\CloudEvent $cloudEvent,
    \Psr\Log\LoggerInterface $logger
    ) {
        $logger->alert('Received event: {event}', ['event' => $cloudEvent]);
        return ['status' => 'SUCCESS'];
    }
);
$app->start();

Common Expression Language (CEL)

In these examples, depending on the type of the event (event.type), the application will be called on /widgets, /gadgets or /products. The expressions are written as Common Expression Language (CEL) where event represents the cloud event. Any of the attributes from the CloudEvents core specification can be referenced in the expression.

Example expressions

Match “important” messages

has(event.data.important) && event.data.important == true

Match deposits greater than $10000

event.type == "deposit" && event.data.amount > 10000

Match multiple versions of a message

event.type == "mymessage.v1"

event.type == "mymessage.v2"

CloudEvent attributes

For reference, the following attributes are from the CloudEvents specification.

Event Data

data

As defined by the term Data, CloudEvents MAY include domain-specific information about the occurrence. When present, this information will be encapsulated within data.

• Description: The event payload. This specification does not place any restriction on the type of this information. It is encoded into a media format which is specified by the datacontenttype attribute (e.g. application/json), and adheres to the dataschema format when those respective attributes are present.
• Constraints:
  • OPTIONAL

Limitation

Currently, it is only possible to access the attributes inside data if it is nested JSON values and not JSON escaped in a string.

REQUIRED Attributes

The following attributes are REQUIRED to be present in all CloudEvents:

id

• Type: String
• Description: Identifies the event. Producers MUST ensure that source + id is unique for each distinct event. If a duplicate event is re-sent (e.g. due to a network error) it MAY have the same id. Consumers MAY assume that Events with identical source and id are duplicates.
• Constraints:
  • REQUIRED
  • MUST be a non-empty string
  • MUST be unique within the scope of the producer
• Examples:
  • An event counter maintained by the producer
  • A UUID

source

• Type: URI-reference

• Description: Identifies the context in which an event happened. Often this will include information such as the type of the event source, the organization publishing the event or the process that produced the event. The exact syntax and semantics behind the data encoded in the URI is defined by the event producer.

  Producers MUST ensure that source + id is unique for each distinct event.

  An application MAY assign a unique source to each distinct producer, which makes it easy to produce unique IDs since no other producer will have the same source. The application MAY use UUIDs, URNs, DNS authorities or an application-specific scheme to create unique source identifiers.

  A source MAY include more than one producer. In that case the producers MUST collaborate to ensure that source + id is unique for each distinct event.

• Constraints:

  • REQUIRED
  • MUST be a non-empty URI-reference
  • An absolute URI is RECOMMENDED

• Examples

  • Internet-wide unique URI with a DNS authority.
    • https://github.com/cloudevents
    • mailto:cncf-wg-serverless@lists.cncf.io
  • Universally-unique URN with a UUID:
    • urn:uuid:6e8bc430-9c3a-11d9-9669-0800200c9a66
  • Application-specific identifiers
    • /cloudevents/spec/pull/123
    • /sensors/tn-1234567/alerts
    • 1-555-123-4567

specversion

• Type: String

• Description: The version of the CloudEvents specification which the event uses. This enables the interpretation of the context. Compliant event producers MUST use a value of 1.0 when referring to this version of the specification.

  Currently, this attribute will only have the ‘major’ and ‘minor’ version numbers included in it. This allows for ‘patch’ changes to the specification to be made without changing this property’s value in the serialization. Note: for ‘release candidate’ releases a suffix might be used for testing purposes.

• Constraints:

  • REQUIRED
  • MUST be a non-empty string

type

• Type: String
• Description: This attribute contains a value describing the type of event related to the originating occurrence. Often this attribute is used for routing, observability, policy enforcement, etc. The format of this is producer defined and might include information such as the version of the type - see Versioning of CloudEvents in the Primer for more information.
• Constraints:
  • REQUIRED
  • MUST be a non-empty string
  • SHOULD be prefixed with a reverse-DNS name. The prefixed domain dictates the organization which defines the semantics of this event type.
• Examples
  • com.github.pull_request.opened
  • com.example.object.deleted.v2

OPTIONAL Attributes

The following attributes are OPTIONAL to appear in CloudEvents. See the Notational Conventions section for more information on the definition of OPTIONAL.

datacontenttype

• Type: String per RFC 2046

• Description: Content type of data value. This attribute enables data to carry any type of content, whereby format and encoding might differ from that of the chosen event format. For example, an event rendered using the JSON envelope format might carry an XML payload in data, and the consumer is informed by this attribute being set to “application/xml”. The rules for how data content is rendered for different datacontenttype values are defined in the event format specifications; for example, the JSON event format defines the relationship in section 3.1.

  For some binary mode protocol bindings, this field is directly mapped to the respective protocol’s content-type metadata property. Normative rules for the binary mode and the content-type metadata mapping can be found in the respective protocol.

  In some event formats the datacontenttype attribute MAY be omitted. For example, if a JSON format event has no datacontenttype attribute, then it is implied that the data is a JSON value conforming to the “application/json” media type. In other words: a JSON-format event with no datacontenttype is exactly equivalent to one with datacontenttype="application/json".

  When translating an event message with no datacontenttype attribute to a different format or protocol binding, the target datacontenttype SHOULD be set explicitly to the implied datacontenttype of the source.

• Constraints:

  • OPTIONAL
  • If present, MUST adhere to the format specified in RFC 2046
• For Media Type examples see IANA Media Types

dataschema

• Type: URI
• Description: Identifies the schema that data adheres to. Incompatible changes to the schema SHOULD be reflected by a different URI. See Versioning of CloudEvents in the Primer for more information.
• Constraints:
  • OPTIONAL
  • If present, MUST be a non-empty URI

subject

• Type: String

• Description: This describes the subject of the event in the context of the event producer (identified by source). In publish-subscribe scenarios, a subscriber will typically subscribe to events emitted by a source, but the source identifier alone might not be sufficient as a qualifier for any specific event if the source context has internal sub-structure.

  Identifying the subject of the event in context metadata (opposed to only in the data payload) is particularly helpful in generic subscription filtering scenarios where middleware is unable to interpret the data content. In the above example, the subscriber might only be interested in blobs with names ending with ‘.jpg’ or ‘.jpeg’ and the subject attribute allows for constructing a simple and efficient string-suffix filter for that subset of events.

• Constraints:

  • OPTIONAL
  • If present, MUST be a non-empty string

• Example:

  • A subscriber might register interest for when new blobs are created inside a blob-storage container. In this case, the event source identifies the subscription scope (storage container), the type identifies the “blob created” event, and the id uniquely identifies the event instance to distinguish separate occurrences of a same-named blob having been created; the name of the newly created blob is carried in subject:
    • source: https://example.com/storage/tenant/container
    • subject: mynewfile.jpg

time

• Type: Timestamp
• Description: Timestamp of when the occurrence happened. If the time of the occurrence cannot be determined then this attribute MAY be set to some other time (such as the current time) by the CloudEvents producer, however all producers for the same source MUST be consistent in this respect. In other words, either they all use the actual time of the occurrence or they all use the same algorithm to determine the value used.
• Constraints:
  • OPTIONAL
  • If present, MUST adhere to the format specified in RFC 3339

Limitation

Currently, comparisons to time (e.g. before or after “now”) are not supported.

Community call demo

Watch this video on how to use message routing with pub/sub:

Next steps

• Try the Pub/Sub routing sample
• Learn about topic scoping
• Learn about message time-to-live
• Learn how to configure Pub/Sub components with multiple namespaces
• List of pub/sub components
• Read the API reference

Last modified February 18, 2022: Update setup-jetstream.md (#2200) (428d8c2)