Send a CloudEvent over NATS¶

Publishing¶

nats.Header is defined as map[string][]string, the same type as envelope.Wire.Headers, so this is a type conversion rather than a copy. NATS has no native content type, so Wire.ContentType goes in as the content-type header:

format := cloudevents.Binary(cloudevents.NATS)

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

header := nats.Header(w.Headers)
if header == nil {
    header = nats.Header{}
}

if w.ContentType != "" {
    header["content-type"] = []string{w.ContentType}
}

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

The nil check is for Structured(), which writes no attribute headers. Binary always returns a map.

Use go/nats's client rather than a bare nats.Conn. These modules have no opinion about which you use, but that one is where the bounded subscriptions and the shed reporting live.

Subscribing¶

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 {
        refused.Add(1)
        logger.Error("refused a CloudEvent", "id", e.ID, "err", err)

        return
    }

    handle(e)
})

Decode takes the content type from the content-type header when Wire.ContentType is empty, whatever the header's case. To accept binary and structured mode on one subject, use envelope.Read instead: Read and write more than one format.

Route on the type, not on the subject¶

A NATS subject and a CloudEvent type say overlapping things, and the temptation is to encode the type into the subject and stop setting the attribute. Do not: the attribute is what makes the event readable by a consumer that is not on NATS, and the round trip through HTTP or an object store keeps it while the subject is lost at the first hop.

Publish to a subject that names the stream, and let type name the occurrence.

What this does not do¶

There is no Send(conn, subject, event) helper here, and there will not be one. It would have to import nats.go, which would put NATS in the dependency graph of every service that sends a CloudEvent over HTTP. TestNoTransportIsImported enforces it, in both go/envelope and go/envelope-cloudevents.