Class: AceMQ::AMQP::AvroCodec

Inherits:
Object
  • Object
show all
Defined in:
lib/acemq/amqp/codec/avro.rb

Overview

Reads and writes Avro.

Compact on the wire and, unlike Protocol Buffers, able to resolve a writer's schema against a reader's — which is what lets a producer add a field without every consumer being redeployed the same afternoon.

There is no codec without a schema, and there cannot be one

Avro's bytes describe nothing about themselves: a reader must already hold the schema the writer used, or the message is unreadable. So this is reached through AvroCodec.of or AvroCodec.registered rather than new, and avro is not a name Codecs knows — a name in configuration cannot carry a schema, and the Java library leaves it out of its own registry for the same reason.

Two modes, and the choice matters more than it looks

AvroCodec.of fixes one schema for the codec's whole life. Small, fast, and nothing extra to run — and the writer's schema is whatever the reader happens to hold. The moment a producer adds a field, every consumer still holding the old schema reads the new bytes wrongly, and Avro will not always notice. Sound only where producer and consumer are released together.

codec = AvroCodec.of(schema_json)

AvroCodec.registered writes the schema's identifier into the front of every message, so a reader can look up exactly what the writer used and let Avro resolve it against its own. This is what makes adding a field safe, and it is the mode to use unless there is a reason not to. The framing is one zero byte, then four bytes of identifier, big-endian, then the Avro body — the layout Confluent's clients use, so messages written here can be read by them, and by the Java, Go and .NET libraries, and the other way round.

registry = Patterns::InMemorySchemaRegistry.new
codec = AvroCodec.registered(registry, subject: "order.placed", schema: schema_json)

Reading against a schema of your own

A registered codec resolves every message onto the schema it holds, and by default that is the same schema it writes with. Where the two want to be different, reader_schema: says so:

codec = AvroCodec.registered(registry, subject: "order.placed",
                           schema: v3_json, reader_schema: v1_json)

The codec then publishes v3 and registers v3 under the subject, while every message it reads — whatever version wrote it — is resolved onto v1. That is the shape a consumer usually wants, and it is not the same as passing v1 as schema:: a codec that both reads and publishes would then register v1 as a new version of the subject and walk the subject backwards. Every library has this: Java spells it registered(registry, readerSchema), .NET ReaderSchema, Go avro.ReadAs(schema) and Python reader_schema=, which is the spelling followed here.

What resolution buys is Avro's, not this library's: a field the writer added that the reader has never heard of is skipped rather than shifting every field after it, and a field the reader expects that the writer never sent is filled in from the reader's own default. A change Avro cannot resolve — a field whose type changed, a field added without a default — raises DecodeError naming both schemas, because no reader schema will make those bytes readable and the sooner that is said the better.

Why each mode claims only its own content type

The two framings are not interchangeable and the difference is invisible in the bytes: a registered message begins with five bytes of framing that a fixed-schema codec reads as the beginning of the first field. That does not throw — Avro decodes the shifted bytes into whatever they happen to mean — so a codec that accepted the other framing would hand back a record full of silent nonsense. Each accepts only its own, which is what the Java and .NET libraries do. The Go library accepts both in either mode, and is the outlier.

Which is why the content type, and not the shape of the bytes, decides what a fixed-schema codec reads: the sender already said which framing it wrote, and guessing from a leading zero byte refuses real messages whose first field encodes to zero. See #unframed for the whole rule.

The gem

avro, required lazily and named when it is missing, so that the gem can go on declaring no runtime dependencies.

Constant Summary collapse

FIXED_CONTENT_TYPE =

What a codec with a fixed schema writes. Java, Go and .NET agree.

"avro/binary"
REGISTERED_CONTENT_TYPE =

What a codec framing a schema identifier writes. Java, Go and .NET agree.

"application/vnd.acemq.avro"
MAGIC =

Confluent's framing: one zero byte, then four bytes of identifier.

0
FRAME_BYTES = 
5

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(schema:, registry: nil, subject: nil, reader_schema: nil) ⇒ AvroCodec

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of AvroCodec.



152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
 
# File 'lib/acemq/amqp/codec/avro.rb', line 152

def initialize(schema:, registry: nil, subject: nil, reader_schema: nil)
  load_runtime!
  @schema = parse(schema)
  # Only a registered codec learns the writer's schema per message, so
  # only a registered codec has two schemas to resolve. A fixed one reads
  # what it writes by definition, and a reader schema there would be a
  # setting that quietly did nothing.
  if reader_schema && registry.nil?
    raise ArgumentError,
          "a reader schema only means something with a registry, which is where the " \
          "writer's schema comes from. Use registered(registry, ...) to read against it."
  end

  @reader_schema = reader_schema.nil? ? @schema : parse(reader_schema)
  @registry = registry
  @subject = subject
  @lock = Mutex.new
  @schema_id = nil
  @by_id = {}
end

Class Method Details

.of(schema) ⇒ AvroCodec

Returns a codec fixed to that schema, carrying nothing on the wire.

Parameters:

  • schema (String, Hash) —

    an Avro schema, as JSON text or as the parsed structure

Returns:

  • (AvroCodec) —

    a codec fixed to that schema, carrying nothing on the wire



122
123
124
 
# File 'lib/acemq/amqp/codec/avro.rb', line 122

def of(schema)
  new(schema: schema)
end

.registered(registry, subject:, schema:, reader_schema: nil) ⇒ AvroCodec

Parameters:

  • registry (#register, #by_id) —

    where schema identifiers are resolved; Patterns::InMemorySchemaRegistry is one

  • subject (String) —

    groups the versions of one message type, conventionally the type itself: "order.placed"

  • schema (String, Hash) —

    the schema this codec writes with, and the one it registers under subject. Also what it reads against, unless reader_schema says otherwise.

  • reader_schema (String, Hash, nil) (defaults to: nil) —

    the schema this consumer was written against. Every message is resolved onto it from whatever the writer used, which is what schema evolution actually needs: a field the reader does not know is skipped, and one the writer omitted is filled in from the reader's default. Left out, schema serves as both, which is what this did before the keyword existed.

Returns:



140
141
142
143
144
145
146
147
148
 
# File 'lib/acemq/amqp/codec/avro.rb', line 140

def registered(registry, subject:, schema:, reader_schema: nil)
  unless registry.respond_to?(:register) && registry.respond_to?(:by_id)
    raise ArgumentError,
          "#{registry.inspect} is not a schema registry: it needs register and by_id"
  end

  new(schema: schema, registry: registry, subject: subject,
      reader_schema: reader_schema)
end

Instance Method Details

#can_decode?(content_type) ⇒ Boolean

Accepts the Avro content types, and only the framing this codec writes.

application/avro and any +...+avro+ media type are taken in either mode, because both are names for Avro without saying which framing, and all four libraries accept them.

Never a message whose sender set no content type: Avro bytes are not recognisable, so volunteering would mean decoding whatever arrived and reporting nonsense as a success.

Returns:

  • (Boolean) 


226
227
228
229
230
231
232
233
 
# File 'lib/acemq/amqp/codec/avro.rb', line 226

def can_decode?(content_type)
  lower = content_type.to_s.downcase
  return false if lower.empty?
  return registered? if lower.start_with?(REGISTERED_CONTENT_TYPE)
  return !registered? if lower.start_with?(FIXED_CONTENT_TYPE, "avro/")

  lower.start_with?("application/avro") || lower.include?("+avro")
end

#content_type ⇒ Object

avro/binary with a fixed schema, application/vnd.acemq.avro with a registry, because the two framings are different messages.



178
 
# File 'lib/acemq/amqp/codec/avro.rb', line 178

def content_type = registered? ? REGISTERED_CONTENT_TYPE : FIXED_CONTENT_TYPE

#decode(body, content_type = nil) ⇒ Object

Returns the datum, resolved onto this codec's reader schema.

Parameters:

  • body (String)
  • content_type (String, nil) (defaults to: nil) —

    what the sender said these bytes are, which is the only reliable way to tell the two framings apart

Returns:

  • (Object) —

    the datum, resolved onto this codec's reader schema

Raises:

  • (DecodeError) —

    when the bytes are not Avro this codec can read, or when the writer's schema cannot be resolved onto the reader's



201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
 
# File 'lib/acemq/amqp/codec/avro.rb', line 201

def decode(body, content_type = nil)
  bytes = body.to_s.b
  writer_schema, offset = registered? ? framed(bytes) : unframed(bytes, content_type)
  # Both schemas go to Avro, which is the whole point: it resolves the
  # difference rather than trusting that there is none.
  reader = Avro::IO::DatumReader.new(writer_schema, @reader_schema)
  reader.read(Avro::IO::BinaryDecoder.new(StringIO.new(bytes[offset..].to_s)))
rescue DecodeError
  raise
rescue StandardError => e
  raise DecodeError, unresolvable(writer_schema, e) if unresolvable?(writer_schema, e)

  raise DecodeError,
        "this message is not Avro that reads as #{reader_schema_name}: #{e.message}"
end

#encode(payload) ⇒ String

Returns the bytes to publish.

Parameters:

  • payload (Hash, Object) —

    a datum matching the schema

Returns:

  • (String) —

    the bytes to publish

Raises:

  • (EncodeError) —

    when the payload does not fit the schema



183
184
185
186
187
188
189
190
191
192
193
 
# File 'lib/acemq/amqp/codec/avro.rb', line 183

def encode(payload)
  out = StringIO.new(+"".b, "wb")
  out.write(frame(schema_id)) if registered?
  Avro::IO::DatumWriter.new(@schema).write(payload, Avro::IO::BinaryEncoder.new(out))
  out.string
rescue EncodeError
  raise
rescue StandardError => e
  raise EncodeError,
        "this payload does not fit the Avro schema #{schema_name}: #{e.message}"
end

#registered? ⇒ Boolean

Whether messages carry a schema identifier.

Returns:

  • (Boolean) 


174
 
# File 'lib/acemq/amqp/codec/avro.rb', line 174

def registered? = !@registry.nil?