CloudEvents
Cloud events is a CNCF specification for describing event data in a common way. CloudEvents seeks to dramatically simplify event declaration and delivery across services, platforms, and beyond. In Occurrent, you don’t persist your domain events directly to an event store, instead you convert them to a cloud event. You may regard a CloudEvent as a standardized envelope around the data in your domain event.
In practice, this means that instead of storing events in a proprietary or arbitrary format, Occurrent, stores events in accordance with the cloud event specification, even at the data-store level. I.e. you know the structure of your events, even in the database that the event store uses. It’s up to you as a user of the library to convert your domain events into cloud events when writing to the event store. This is extremely powerful, not only does it allow you to design your domains event in any way you find fit (for example without compromises enforced by a JSON serialization library) but it also allows for easier migration, data consistency and features such as (fully-consistent) queries to the event store for certain use cases. A cloud event is made-up by a set of pre-defined attributes described in the cloud event specification. In the context of event sourcing, we can leverage these attributes in the way suggested below:
| Cloud Event Attribute Name | Event Sourcing Nomenclature | Description |
|---|---|---|
| id | event id | The cloud event id attribute is used to store the id of a unique event in a particular context (“source”). Note that this id doesn’t necessarily need to be globally unique (but the combination of id and source must). Typically this would be a UUID. |
| source | category | You can regard the “source” attribute as the “stream type” or a “category” for certain streams. For example, if you’re creating a game, you may have two kinds of aggregates in your bounded context, a “game” and a “player”. You can regard these as two different sources (categories). These are represented as URN’s, for example the “game” may have the source “urn:mycompany:mygame:game” and “player” may have “urn:mycompany:mygame:player”. This allows, for example, subscriptions to subscribe to all events related to any player (by using a subscription filter for the source attribute). |
| subject | “subject” (~identifier) | A subject describes the event in the context of the source, typically an entity (aggregate) id that all events in the stream are related to. This property is optional (because Occurrent automatically adds the streamid attribute) and it’s possible that you may not need to add it. But it can be quite useful. For example, a stream may not necessarily, just hold contents of a single aggregate, and if so the subject can be used to distinguish between different aggregates/entities in a stream. Another example would be if you have multiple streams that represents different aspects of the same entity. For example, if you have a game where players are awarded points based on their performance in the game after the game has ended, you may decide to represent “point awarding” and “game play” as different streams, but they refer to the same “game id”. You can then use the “game id” as subject. |
| type | event type | The type of the event. It may be enough to just use name of the domain event, such as “GameStarted” but you may also consider using a URN (e.g. “urn:mycompany:game:started”) or qualify it (“com.mycompany.game.started”). Note that you should try to avoid using the fully-qualified class name of the domain event since you’ll run into trouble if you’re moving the domain event to a different package. |
| time | event time | The time when the event occurred (typically would be the application time and not the processing time) described by RFC 3339 (represented as java.time.OffsetDateTime by the CloudEvent SDK). |
| datacontenttype | content-type | The content-type of the data attribute, typically you want to use “application/json”, which is also the default if you don’t specify any content-type at all. |
| dataschema | schema | The URI to a schema describing the data in the cloud event (optional). |
| data | event data | The actual data needed to represent your domain event, for example the contents of a GameStarted event. You can leave out this attribute entirely if your event is fully described by other attributes. |
Note that the table above is to be regarded as a rule of thumb, it’s ok to map things differently if it’s better suited for your application, but it’s a good idea to keep things consistent throughout your organization. To see an example of how this may look in code, refer to the application service documentation.
Occurrent CloudEvent Extensions
Section titled “Occurrent CloudEvent Extensions”Occurrent automatically adds two extension attributes to each cloud event written to the event store:
| Attribute Name | Type | Description |
|---|---|---|
streamid | String | An id that uniquely identifies a particular event stream. It’s used to determine which events belong to which stream. |
streamversion | Long | The id of the stream version for a particular event. It’s used for optimistic concurrency control. |
These are required for Occurrent to operate. A long-term goal of Occurrent is to come up with a standardized set of cloud event extensions that are agreed upon and used by several different vendors.
Since version 0.30.0, Occurrent also stamps every event with a global, monotonically increasing position extension, shared by stream-written and DCB events alike. It gives stream consumers the same total-ordering guarantee that DCB relies on, and stream catch-up reconciles on it rather than on wall-clock time. It is enabled by default for a new store. When a MongoDB store starts up against a collection that already holds events without a position, and position was only on by default rather than turned on explicitly, the store turns position off for itself instead of building the position index over the existing collection, and logs how to enable it. Enabling position explicitly with EventStoreConfig.withStreamPosition() (or the property occurrent.event-store.stream.position=true) keeps it on even over that existing data, in which case backfill the existing events first. A stream-only store can opt out with EventStoreConfig.withoutStreamPosition() (or occurrent.event-store.stream.position=false), but a store with the DCB capability always has it enabled.
In the meantime, it’s quite possible that Occurrent will provide a wider set of optional extensions in the future (such as correlation id and/or sequence number). But for now, it’s up to you as a user to add these if you need them (see CloudEvent Metadata), you would typically do this by creating or extending/wrapping an already existing application service.
CloudEvent Metadata
Section titled “CloudEvent Metadata”You can specify metadata to the cloud event by making use of extension attributes. This is the place to add things such as sequence number, correlation id, causation id etc. Actually there’s already a standard way of applying distributed tracing and sequence number generation extensions to cloud events that might be of interest.