Events: CloudEvent and EventPublisher

Publish CloudEvents to each cloud's event router: Amazon EventBridge, Azure Event Grid namespaces and Google Cloud Eventarc Advanced, plus an in-memory bus. Routers deliver events to queues by rules you define, and you receive them with an ordinary subscription, decoding each message with CloudEvents.decode. Delivery is at least once, so deduplicate by source and id.

Buses, rules, subscriptions, enrollments and pipelines are infrastructure: create them with your usual tooling.

On Google Cloud, give every event data: Pub/Sub refuses an Eventarc pipeline's delivery of an event without data, so the event is accepted by the bus but never reaches the topic. A pipeline also publishes to Pub/Sub only with an OAuth token: name a service account that can publish to the topic in the destination's authentication, and give the Eventarc service agent roles/iam.serviceAccountTokenCreator on it.

Events or messaging? Messaging sends work to a queue you know. Events announce what happened; the router decides who hears it, and new consumers subscribe without the publisher changing. Events from the cloud itself, such as EC2 state changes, arrive through the same rules.

Events#

CloudEvent placed = CloudEvent.builder("com.acme.order.placed", "/orders")   // type, source
        .subject("o-42")
        .json("{\"total\":12.5}")             // or data(contentType, bytes or text)
        .extension("tenant", "acme")          // 1-20 lower-case letters and digits
        .build();

id defaults to a random UUID and time to now. Events compare by value. Data is JSON, text or bytes with its media type, or absent. CloudEvents.toJson and CloudEvents.fromJson convert to and from the CloudEvents JSON format.

Publishing#

EventPublisher orders = halyx.provider("orders").eventPublisher("orders");   // the bus

orders.publish(placed);
orders.publishAll(List.of(placed, shipped));   // in as few requests as the router allows

Events larger than the router's limit are refused with InvalidArgumentException before anything is sent. publishAll isn't atomic: when a request fails, earlier requests' events stay published. Retrying is safe, since each event keeps its id. Publishing goes through the provider's interceptors under the service name events, so retries and metrics apply as they do to other services.

EventPublisher grid = azure.eventPublisher(EventBusConfig.builder("orders")
        .property(AzureProperties.EVENT_GRID_ENDPOINT, "https://acme.australiaeast-1.eventgrid.azure.net")
        .build());

On EventBridge the event's source and type become the entry's Source and DetailType, so rules can match them, and the whole CloudEvent travels as detail.

Without blocking#

async() gives the publisher's asynchronous view: publish and publishAll as CompletableFutures, with the same size checks (every event is checked before any is sent) and errors. They're non-blocking on EventBridge, Event Grid and Eventarc; with Micrometer Tracing, the trace is taken from the calling thread.

orders.async().publish(placed).thenRun(() -> log.info("published {}", placed.id()));

Receiving#

Receive from the queue the router delivers to, and decode each message:

Subscription inbox = halyx.provider("orders").subscription("orders-inbox");   // SQS queue

try (MessageConsumer consumer = inbox.listen(message -> {
    CloudEvent event = CloudEvents.decode(message);
    // ...
    message.ack();
}, ConsumerOptions.DEFAULTS)) {
    // ...
}

CloudEvents.decode reads every form the routers deliver:

Messages that hold none of these throw IllegalArgumentException. Routers may reformat JSON data (EventBridge changes whitespace and number formatting), so compare JSON data as JSON.

An Event Grid pull subscription is a subscription with the namespace endpoint and its topic:

Subscription inbox = azure.subscription(SubscriptionConfig.builder("billing").topic("orders")
        .property(AzureProperties.EVENT_GRID_ENDPOINT, "https://acme.australiaeast-1.eventgrid.azure.net")
        .property(AzureProperties.EVENT_GRID_LOCK_DURATION, "PT60S")   // as the subscription is configured
        .build());

It supports ack, nack with a delay, and lease extension; receive waits between 10 and 120 seconds.

In Spring Boot#

Event buses become EventPublisher beans, and @HalyxListener methods can take a CloudEvent. See Spring Boot.