Skip to content

CloudEvents: getting started

By the end of this you will have sent a CloudEvent between two halves of one program and read every attribute back at the other end. It needs go/nats for the transport, and nothing else.

1. Install

$ go get gitlab.com/phpboyscout/go/envelope
$ go get gitlab.com/phpboyscout/go/envelope-cloudevents

The first is the event and the Format seam; the second is CloudEvents' two formats. The second package is named cloudevents, so import it by that name:

import (
    "gitlab.com/phpboyscout/go/envelope"
    cloudevents "gitlab.com/phpboyscout/go/envelope-cloudevents"
)

2. Build an event

func order() envelope.Event {
    return envelope.Event{
        ID:              "01JQ8Z3F7K2M4N6P8R0T2V4W6X",
        Source:          "/phpbotscout/ingest",
        Type:            "uk.phpboyscout.orders.created",
        Subject:         "order-4172",
        Time:            time.Now().UTC(),
        DataContentType: "application/json",
        Data:            []byte(`{"order":"4172"}`),
    }
}

SpecVersion is deliberately absent. There is exactly one value the formats can write, so Encode fills it in rather than making every caller repeat it.

ID and Source together identify the occurrence, and a producer must not reuse the pair. A ULID or a UUID is the usual choice.

3. Encode it

format := cloudevents.Binary(cloudevents.NATS)

w, err := format.Encode(order())
if err != nil {
    return err
}

w is an envelope.Wire. Its Headers field is a map[string][]string holding six keys: ce-id, ce-source, ce-specversion, ce-type, ce-subject and ce-time. The content type is in w.ContentType (application/json), not in a header, and w.Body is the payload untouched. Values are percent-encoded, so read them back with Decode rather than by hand.

http.Header.Get will not find them

Keys are written in lower case, which is the NATS binding's form. http.Header.Get canonicalises the key it is given, so hdr.Get("Ce-Id") returns "" against this map. Read it by iteration or by exact key, and let net/http canonicalise on the wire.

Encode returns an error rather than producing something no consumer would accept, so a missing Type is caught in your process rather than in somebody else's subscriber.

4. Send it

nats.Header is map[string][]string, so the conversion is a conversion and not a copy. NATS has no native content type, so it goes in as the content-type header:

header := nats.Header(w.Headers)
if w.ContentType != "" {
    header["content-type"] = []string{w.ContentType}
}

err = client.PublishMsg(ctx, &nats.Msg{
    Subject: "orders.created",
    Header:  header,
    Data:    w.Body,
})
if err != nil {
    return err
}

client is a go/nats client. ContentType is empty when the event has no DataContentType, so the header is only set when there is something to put in it.

5. Read it at the other end

sub, err := client.Subscribe("orders.created", func(msg *nats.Msg) {
    e, _, err := format.Decode(envelope.Wire{Headers: msg.Header, Body: msg.Data})
    if err != nil {
        // e still carries whatever could be read: log e.ID, not "a message failed".
        logger.Error("refused a CloudEvent", "id", e.ID, "source", e.Source, "err", err)

        return
    }

    logger.Info("received", "type", e.Type, "subject", e.Subject)
})

Decode reads the content type from the content-type header here, because Wire.ContentType is empty. The second return value, the envelope.Foreign, holds every header that is not part of the event, such as Nats-Msg-Id; it is ignored above.

The error branch is the part worth pausing on. Decode refuses a malformed event, and hands back the attributes it did manage to read, because core NATS has no dead letter, so a dropped message is gone unless something recorded which one it was. See Handle a refused event.

Where next