Codecs

A codec turns a message payload into bytes and back.

type Codec interface {
    ContentType() string
    Encode(payload any) ([]byte, error)
    Decode(body []byte, dst any) error
    CanDecode(contentType string) bool
}

Decode takes a pointer to the destination rather than returning a value, which is how encoding/json and every other Go decoder works. It is the one place this API deliberately departs from the Java and .NET shape: returning an any and asking the caller to type-assert would be worse Go for no gain.

JSON

The default, and the only one the core package carries — it needs nothing outside the standard library.

mq, err := acemq.Connect(ctx, url) // JSONCodec unless told otherwise

Tag your fields

type OrderPlaced struct {
    OrderID    string `json:"orderId"`
    TotalCents int64  `json:"totalCents"`
}

Unlike the Java and .NET codecs there is no camelCase policy to configure, because Go already puts the wire name in the tag. The corollary is that an untagged field goes on the wire under its Go name — OrderID, capitals and all — and nothing in another language will read it. Tag the fields.

It answers for a message with no content type

CanDecode("") returns true. JSON is the default format, an untyped message is far more likely to be JSON than anything else, and something has to read it.

This is the opposite of the choice the YAML and TOML codecs make in the .NET library, where answering for an untyped message would record traffic under a format nobody sent. Being the default is what earns JSON the benefit of the doubt.

A malformed body is fatal

if err := json.Unmarshal(body, dst); err != nil {
    return Fatalf("acemq: this message is not JSON that reads as %T: %w", dst, err)
}

The same bytes fail the same way every time, so the message is dead-lettered rather than retried until it ages out. It never reaches the handler. See retries.

Choosing one

Per connection:

mq, err := acemq.Connect(ctx, url, acemq.WithCodec(myCodec))

Per publisher or per consumer, so one connection can send JSON to one queue and something denser to another:

pub := acemq.NewPublisher[Telemetry](mq, "", "telemetry",
    acemq.PublishWith[Telemetry](myCodec))

sub, err := acemq.Consume(ctx, mq, "telemetry", handler,
    acemq.ConsumeWith(myCodec))

Writing one

type CSVCodec struct{}

func (CSVCodec) ContentType() string { return "text/csv" }

func (CSVCodec) Encode(payload any) ([]byte, error) {
    row, ok := payload.(Row)
    if !ok {
        return nil, acemq.Fatalf("acemq: CSVCodec cannot encode a %T", payload)
    }
    return []byte(row.String()), nil
}

func (CSVCodec) Decode(body []byte, dst any) error {
    row, ok := dst.(*Row)
    if !ok {
        return acemq.Fatalf("acemq: CSVCodec cannot decode into a %T", dst)
    }
    parsed, err := ParseRow(string(body))
    if err != nil {
        return acemq.Fatalf("acemq: this message is not a CSV row: %w", err)
    }
    *row = parsed
    return nil
}

func (CSVCodec) CanDecode(contentType string) bool {
    return strings.HasPrefix(contentType, "text/csv")
}

Two things worth getting right:

Mark decoding failures fatal. A body that will not decode will not decode next time either. Without the mark it is retried until the attempts run out, which delays the dead-lettering that was always going to happen and holds a prefetch slot while it does.

Decide what CanDecode("") should mean. Answering for a message whose sender set no content type is right for a default format and wrong for a specialised one — it records traffic as a format nobody sent, and the mistake surfaces much later.

The registry

Configuration can name a format without the calling code importing it:

acemq.RegisterCodec("csv", func() acemq.Codec { return CSVCodec{} })

codec, err := acemq.CodecByName(os.Getenv("ACEMQ_CODEC"))
if err != nil {
    return err
}
mq, err := acemq.Connect(ctx, url, acemq.WithCodec(codec))

json, bytes, and text (also registered as string) come with the package. text and string are the same StringCodec — UTF-8 text written as text/plain; charset=utf-8 — under both names the AceMQ libraries use for it: Java and Python call it text, .NET and Ruby string, so one configuration value works across all five. A body that is not valid UTF-8 is refused as fatal rather than decoded into replacement characters, so it is parked.

An unknown name lists what is available, because somebody who typoed one cannot see the registry:

acemq: no codec named "jsn" is registered; known: [bytes csv json string text]

The other formats

Every format the Java and .NET libraries have. Each is a module of its own, so the core keeps its single dependency and an application takes only what it sends:

go get github.com/AceMQ-Company/acemq-go-amqp/codec/yaml

None of them answers for a message with no content type. JSON is the default and gets that benefit of the doubt; the others would be recording traffic under a format nobody sent.

YAML and TOML

Both are for a message a person reads and edits. YAML writes block style, which is the only reason to pay for it over JSON. TOML has no Norway problem: in YAML country: NO is the boolean false, while in TOML an unquoted NO is a parse error rather than a country that quietly became false.

TOML is a table format, so a body must be a struct or a map at the top level. The encoder underneath will happily write a bare list as ["a", "b"] — which is not a TOML document — so this codec refuses it before it goes out rather than letting it fail in the consumer with no clue where it came from.

Avro, and schema evolution

Two modes. avro.Of(schema) carries nothing on the wire and expects both ends to hold the same schema: smallest, and most brittle.

codec, err := avro.Registered(registry, "order.placed", schema)

Registered frames each message with a schema identifier the way Confluent's clients do — one zero byte, four bytes of identifier, big-endian — which is what lets Confluent's clients, the Java library and this one read each other. A producer can then add a field with a default and a consumer that has never heard of it still reads the message. There is a test for exactly that.

Reading against a schema of your own

Looking the writer's schema up is half of what evolution needs. On its own it means a consumer decodes whatever shape the producer sent: a field it has never heard of arrives, and a field it expects is simply absent — read back as the zero value — until the producer starts sending it. ReaderSchema supplies the other half, the schema this consumer was written against:

consumer, err := avro.Registered(registry, "order.placed", schema,
    avro.ReaderSchema(schema))

Avro is then given both schemas and resolves them. A field the writer added and this reader does not know is skipped rather than shifting every field after it, and a field the writer omitted is filled in from the reader's own default — "public", not "". The consumer sees the shape it was written against, whichever version wrote the message.

Passing the codec's own schema, as above, is the ordinary case: a consumer reads what it was compiled against. The two are separate arguments because they are separate schemas — the first is what the codec writes, the second what it reads — and a process that both publishes and consumes may want them to differ.

Resolution happens once per writer schema and is remembered, so the cost lands on the first message carrying an identifier and not again. A writer schema that cannot be resolved onto the reader's — a field added without a default, a type changed to one Avro will not promote — is an error naming both schemas rather than a decode that returns the wrong values. Both are printed in full, because the two are usually versions of one record and share a name.

ReaderSchema changes nothing on the wire. It is a read-side decision: the bytes a producer writes are identical with it and without it, so Java, .NET, Python and Ruby go on reading them.

The name is the family's. Java calls it readerSchema, Python reader_schema, Ruby reader_schema: and .NET readerSchema, so one idea has one spelling in all five and a reader moving between them is not learning a synonym.

The other libraries reach the same behaviour by routes of their own, and the routes are worth knowing when messages cross between them: schema resolution below has the table, and what each behaviour decodes the same bytes to. Why this one is opt-in: a registered codec here has always decoded against the writer's schema, and turning resolution on by default would change what every existing consumer sees — a field it had been ignoring starts arriving as a default, silently. Opting in is one argument; a change of meaning under an unchanged call is not.

The two modes are not interchangeable, and a codec only claims the content type its own mode can read:

Both take application/avro and any *+avro suffix type, because neither says how the body was framed and a message nothing claims is a message nobody reads.

The gate matters. A fixed-schema codec handed a registry-framed message would pass the five framing bytes to the Avro reader as the start of the first field, and Avro does not object: it reads the shifted bytes as whatever they happen to mean and returns a record where every value is wrong, with no error and no log line. Refusing the content type turns silent corruption into a message nothing claimed, which is a thing somebody can see. The other direction was never silent — the registered decoder checks the framing byte and the schema identifier, and says which mode wrote the message it cannot read. Java, .NET, Python and Ruby gate the same way.

Encryption

keyring, err := crypto.NewKeyring(crypto.Key{ID: "2026-01", Secret: secret})
codec := crypto.Wrap(acemq.JSONCodec{}, keyring)

The payload is encoded as usual and the bytes are then encrypted with AES-GCM, so the broker, its disk, its backups and its management interface see ciphertext. GCM authenticates as well as encrypts, so a body altered in the broker fails to open rather than decrypting into something else.

Headers travel in the clear. The envelope is how the library routes and retries, so it cannot be encrypted without the broker losing the ability to do its job. Do not put anything secret in a header.

A keyring holds more than one key because rotation needs an overlap: add the new key everywhere first so every consumer can read it, then make it current. A keyring with one key cannot rotate without an outage.

A key is 16, 24 or 32 bytes — the three lengths AES takes, and the same three Java, Python and Ruby accept. Anything else is refused rather than padded or hashed into shape, because both would make a weak key look like a strong one.

The framing is the family's byte for byte — 0xAE, a version byte, a one-byte key id length, the key id, a 12-byte nonce, then the ciphertext and its 16-byte tag — so a producer in Java, .NET, Python or Ruby and a Go consumer can share a key. It is the only framing this library writes and the only one it reads; a body that does not begin 0xAE is refused as what it is. See security.

Schema resolution

Handed the writer's schema and the reader's, Avro reconciles the two: a field the writer added that the reader does not declare is skipped, and a field the writer omitted is filled in from the reader's default. Handed only the writer's, there is nothing to reconcile, and the record arrives in the shape it was written.

So there is one rule, and it is the same rule in all five AceMQ libraries:

Resolution happens when the library has a reader schema to resolve onto.

What differs between the languages is where a reader schema comes from, and therefore how often there is one. Nothing about the bytes differs.

This is not an inconsistency waiting to be flattened. A library that resolves and a library that does not are both right about the same bytes — they are answering different questions, because only one of them was told what the reader expects.

The case that bites is a field the writer removed that the reader declares with a default. With resolution, the field arrives carrying that default. Without it, the field is simply absent: a missing key, whatever the language calls one. A consumer written against the reader schema then reads a value that was never on the wire, or fails to read a field it is sure it declared, and which of those happens is decided entirely by whether a reader schema was in play.

The other direction is the one people expect to be dangerous and is not. A field the writer added that the reader does not declare is skipped under resolution and present without it, and either way the fields the reader does declare come back correct — the unknown field does not shift the ones after it.

Both cases are pinned, with the bytes, in internal/testdata/avro-resolution-fixtures.json, which every AceMQ library carries a copy of. It records the decoded value under each behaviour, as resolved and writerShape, and which library lands on which.

Asking for resolution in Go

// Resolves: every message is reconciled with the schema this consumer holds.
codec, err := avro.Registered(registry, "order.placed", schema,
    avro.ReaderSchema(schema))

// Does not resolve: the record arrives in the shape the producer wrote it.
asWritten, err := avro.Registered(registry, "order.placed", schema)

Reach for the first unless there is a reason not to. Being able to redeploy a producer without its consumers is the whole point of putting a schema id on the front of the message, and resolution is the half of that which happens on the read side.

One edge worth knowing before you meet it: without a reader schema the removed field is absent, and a Go struct has no way to be absent. Decode that case's body into a struct with a Currency field and it reads back "" — which is also what a producer sending an empty string looks like. Decoding into a map[string]any is what tells the two apart, because there the key is simply not present. With avro.ReaderSchema the question does not arise: the field arrives carrying the reader's default.

Several formats on one queue

codec := acemq.NewCompositeCodec(acemq.JSONCodec{}, yaml.Codec{})

The first is what it writes; all of them are offered a message to read, and the first that claims the content type does. For a migration, or a queue several producers write to in different formats.