Class: AceMQ::AMQP::Patterns::InMemorySchemaRegistry

Inherits:
Object
  • Object
show all
Defined in:
lib/acemq/amqp/patterns/schema.rb

Overview

Message shapes, kept in this process.

A registry is anything answering register, by_id, latest and versions. This one is for tests and for a single service that wants the shape of the thing. It is not a registry in the sense that matters: nothing is shared between processes, so a consumer cannot look up a schema a producer registered elsewhere, which is the entire point of having one. Use a database-backed registry, or Confluent's, for anything real — the wire framing this library uses is compatible with theirs.

Nothing here puts anything on the wire. Which header carries a schema identifier is a cross-language contract, and it is not one AceMQ has agreed yet; a header invented here would be one the Java, Go, .NET and Python libraries could not read, which is worse than none.

Instance Method Summary collapse

Constructor Details

#initialize ⇒ InMemorySchemaRegistry

Returns a new instance of InMemorySchemaRegistry.



60
61
62
63
64
65
66
 
# File 'lib/acemq/amqp/patterns/schema.rb', line 60

def initialize
  @lock = Mutex.new
  @by_id = {}
  @by_subject = Hash.new { |subjects, name| subjects[name] = [] }
  @by_fingerprint = {}
  @next_id = 1
end

Instance Method Details

#by_id(id) ⇒ Object

The schema an identifier names.

Raises:



96
97
98
99
 
# File 'lib/acemq/amqp/patterns/schema.rb', line 96

def by_id(id)
  @lock.synchronize { @by_id[id] } ||
    raise(SchemaNotFound, "no schema with id #{id}")
end

#latest(subject) ⇒ Object

The newest version of a subject.

Raises:



104
105
106
107
 
# File 'lib/acemq/amqp/patterns/schema.rb', line 104

def latest(subject)
  @lock.synchronize { @by_subject[subject].last } ||
    raise(SchemaNotFound, "no schema for subject #{subject.inspect}")
end

#register(subject, format, definition) ⇒ SchemaDefinition

Records a schema and returns it with an identifier.

The same definition registered twice returns the same identifier rather than making a second version. Without that, a service that registers its schemas on every start adds a version per restart, and a week later the subject has three hundred identical versions.

Parameters:

  • subject (String) —

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

  • format (String) —

    "avro", "protobuf", "json-schema" — whatever the schema is written in. Nothing here interprets it.

  • definition (String) —

    the schema

Returns:



81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/acemq/amqp/patterns/schema.rb', line 81

def register(subject, format, definition)
  if subject.to_s.empty? || definition.to_s.empty?
    raise ArgumentError, "a schema needs a subject and a definition"
  end

  fingerprint = Patterns.fingerprint(definition)
  @lock.synchronize do
    @by_fingerprint["#{subject}/#{fingerprint}"] ||= add(subject, format, definition,
                                                         fingerprint)
  end
end

#versions(subject) ⇒ Object

Every version of a subject, oldest first. Empty when there are none, because "what versions are there" has a sensible answer for a subject nobody has registered and "which schema is this" does not.



112
113
114
 
# File 'lib/acemq/amqp/patterns/schema.rb', line 112

def versions(subject)
  @lock.synchronize { @by_subject[subject].dup }
end