Publishing & subscribing messages with Cloudevents

Learn why Dapr uses CloudEvents, how they work in Dapr pub/sub, and how to create CloudEvents.

To enable message routing and provide additional context with each message, Dapr uses the CloudEvents 1.0 specification as its message format. Any message sent by an application to a topic using Dapr is automatically wrapped in a CloudEvents envelope, using the Content-Type header value for datacontenttype attribute.

Dapr uses CloudEvents to provide additional context to the event payload, enabling features like:

  • Tracing
  • Content-type for proper deserialization of event data
  • Verification of sender application

You can choose any of three methods for publish a CloudEvent via pub/sub:

  1. Send a pub/sub event, which is then wrapped by Dapr in a CloudEvent envelope.
  2. Replace specific CloudEvents attributes provided by Dapr by overriding the standard CloudEvent properties.
  3. Write your own CloudEvent envelope as part of the pub/sub event.

Dapr-generated CloudEvents example

Sending a publish operation to Dapr automatically wraps it in a CloudEvent envelope containing the following fields:

  • id
  • source
  • specversion
  • type
  • traceparent
  • traceid
  • tracestate
  • topic
  • pubsubname
  • time
  • datacontenttype (optional)

The following example demonstrates a CloudEvent generated by Dapr for a publish operation to the orders topic that includes:

  • A W3C traceid unique to the message
  • The data and the fields for the CloudEvent where the data content is serialized as JSON
  1. {
  2.   "topic": "orders",
  3.   "pubsubname": "order_pub_sub",
  4.   "traceid": "00-113ad9c4e42b27583ae98ba698d54255-e3743e35ff56f219-01",
  5.   "tracestate": "",
  6.   "data": {
  7.     "orderId": 1
  8.   },
  9.   "id": "5929aaac-a5e2-4ca1-859c-edfe73f11565",
  10.   "specversion": "1.0",
  11.   "datacontenttype": "application/json; charset=utf-8",
  12.   "source": "checkout",
  13.   "type": "com.dapr.event.sent",
  14.   "time": "2020-09-23T06:23:21Z",
  15.   "traceparent": "00-113ad9c4e42b27583ae98ba698d54255-e3743e35ff56f219-01"
  16. }

As another example of a v1.0 CloudEvent, the following shows data as XML content in a CloudEvent message serialized as JSON:

  1. {
  2.   "topic": "orders",
  3.   "pubsubname": "order_pub_sub",
  4.   "traceid": "00-113ad9c4e42b27583ae98ba698d54255-e3743e35ff56f219-01",
  5.   "tracestate": "",
  6.   "data" : "<note><to></to><from>user2</from><message>Order</message></note>",
  7.   "id" : "id-1234-5678-9101",
  8.   "specversion" : "1.0",
  9.   "datacontenttype" : "text/xml",
  10.   "subject" : "Test XML Message",
  11.   "source" : "https://example.com/message",
  12.   "type" : "xml.message",
  13.    "time" : "2020-09-23T06:23:21Z"
  14. }

Replace Dapr generated CloudEvents values

Dapr automatically generates several CloudEvent properties. You can replace these generated CloudEvent properties by providing the following optional metadata key/value:

  • cloudevent.id: overrides id
  • cloudevent.source: overrides source
  • cloudevent.type: overrides type
  • cloudevent.traceid: overrides traceid
  • cloudevent.tracestate: overrides tracestate
  • cloudevent.traceparent: overrides traceparent

The ability to replace CloudEvents properties using these metadata properties applies to all pub/sub components.

Example

For example, to replace the source and id values from the CloudEvent example above in code:

  1. with DaprClient() as client:
  2.     order = {'orderId': i}
  3.     # Publish an event/message using Dapr PubSub
  4.     result = client.publish_event(
  5.         pubsub_name='order_pub_sub',
  6.         topic_name='orders',
  7.         publish_metadata={'cloudevent.id': 'd99b228f-6c73-4e78-8c4d-3f80a043d317', 'cloudevent.source': 'payment'}
  8.     )
  1. var order = new Order(i);
  2. using var client = new DaprClientBuilder().Build();
  4. // Override cloudevent metadata
  5. var metadata = new Dictionary<string,string>() {
  6.     { "cloudevent.source", "payment" },
  7.     { "cloudevent.id", "d99b228f-6c73-4e78-8c4d-3f80a043d317" }
  8. }
  10. // Publish an event/message using Dapr PubSub
  11. await client.PublishEventAsync("order_pub_sub", "orders", order, metadata);
  12. Console.WriteLine("Published data: " + order);
  14. await Task.Delay(TimeSpan.FromSeconds(1));

The JSON payload then reflects the new source and id values:

  1. {
  2.   "topic": "orders",
  3.   "pubsubname": "order_pub_sub",
  4.   "traceid": "00-113ad9c4e42b27583ae98ba698d54255-e3743e35ff56f219-01",
  5.   "tracestate": "",
  6.   "data": {
  7.     "orderId": 1
  8.   },
  9.   "id": "d99b228f-6c73-4e78-8c4d-3f80a043d317",
  10.   "specversion": "1.0",
  11.   "datacontenttype": "application/json; charset=utf-8",
  12.   "source": "payment",
  13.   "type": "com.dapr.event.sent",
  14.   "time": "2020-09-23T06:23:21Z",
  15.   "traceparent": "00-113ad9c4e42b27583ae98ba698d54255-e3743e35ff56f219-01"
  16. }

Important

While you can replace traceid/traceparent and tracestate, doing this may interfere with tracing events and report inconsistent results in tracing tools. It’s recommended to use Open Telementry for distributed traces. Learn more about distributed tracing.

Publish your own CloudEvent

If you want to use your own CloudEvent, make sure to specify the datacontenttype as application/cloudevents+json.

If the CloudEvent that was authored by the app does not contain the minimum required fields in the CloudEvent specification, the message is rejected. Dapr adds the following fields to the CloudEvent if they are missing:

  • time
  • traceid
  • traceparent
  • tracestate
  • topic
  • pubsubname
  • source
  • type
  • specversion

You can add additional fields to a custom CloudEvent that are not part of the official CloudEvent specification. Dapr will pass these fields as-is.

Example

Publish a CloudEvent to the orders topic:

  1. dapr publish --publish-app-id orderprocessing --pubsub order-pub-sub --topic orders --data '{\"orderId\": \"100\"}'

Publish a CloudEvent to the orders topic:

  1. curl -X POST http://localhost:3601/v1.0/publish/order-pub-sub/orders -H "Content-Type: application/cloudevents+json" -d '{"specversion" : "1.0", "type" : "com.dapr.cloudevent.sent", "source" : "testcloudeventspubsub", "subject" : "Cloud Events Test", "id" : "someCloudEventId", "time" : "2021-08-02T09:00:00Z", "datacontenttype" : "application/cloudevents+json", "data" : {"orderId": "100"}}'

Publish a CloudEvent to the orders topic:

  1. Invoke-RestMethod -Method Post -ContentType 'application/cloudevents+json' -Body '{"specversion" : "1.0", "type" : "com.dapr.cloudevent.sent", "source" : "testcloudeventspubsub", "subject" : "Cloud Events Test", "id" : "someCloudEventId", "time" : "2021-08-02T09:00:00Z", "datacontenttype" : "application/cloudevents+json", "data" : {"orderId": "100"}}' -Uri 'http://localhost:3601/v1.0/publish/order-pub-sub/orders'

Event deduplication

When using cloud events created by Dapr, the envelope contains an id field which can be used by the app to perform message deduplication. Dapr does not handle deduplication automatically. Dapr supports using message brokers that natively enable message deduplication.

Next steps

Last modified October 12, 2023: Update config.toml (#3826) (0ffc2e7)