Class: Sourced::Message::Codec

Inherits:
Object
  • Object
show all
Defined in:
lib/sourced/message/codec.rb

Overview

Compiles a [decoder, encoder] Pair per message class and serializes messages through it. Abstract: a subclass binds the wire format by answering Codec.default_format. JSONCodec and FormsCodec are the two that ship.

codec = Sourced::Message::JSONCodec.default.compile!
codec.encode(message)  # => structures native to the format
codec.decode(attrs)    # => the message

Pairs are keyed by type string and built by #compile! from a Registry; a type missing from the result raises, so a type defined after the compile stays invisible until a fresh codec is built. Compiling is always explicit — nothing here compiles itself on first use — which makes "when is serialization settled?" a question with one answer per process.

It encodes the whole message, envelope included, which suits a transport carrying a message as a single document (a file, a socket frame, a form submission). A store that keeps the envelope in columns wants only the payload encoded; Store::MessageCodec subclasses JSONCodec for exactly that. The three private seams below are the whole subclass contract:

#compiled_type(klass)           what Plumb type to compile for a message class
#encode_subject(message)        what #encode feeds the encoder
#build(klass, attrs, decoder)   what #decode returns

Every subclass gets its own .default and .pairs for free (both are plain class ivars), which is load-bearing: one message class compiles to a different pair under each format and each seam set, and they must not collide.

Direct Known Subclasses

FormsCodec, JSONCodec

Defined Under Namespace

Classes: Pair

Constant Summary collapse

EncodeError =

Raised when a message being written can't be represented in the codec's format, which in practice means the message itself is invalid.

Class.new(StandardError)
DecodeError =

Raised when a serialized message no longer satisfies its class's schema — a schema change, a hand-edited record, a foreign writer.

Class.new(StandardError)
UnregisteredTypeError =

Raised when asked for a type this codec has no pair for: one defined after the compile, one absent from this process, or any type at all when nothing has compiled yet.

Class.new(StandardError)

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(format: self.class.default_format, registry: Sourced::Message.registry) ⇒ Codec

Returns a new instance of Codec.

Parameters:

  • format (Class<Plumb::Codec>) (defaults to: self.class.default_format) —

    the format compiled onto message types. Defaults to the subclass's default_format; pass one to scope a codec to its own set of encoders, as specs do.

  • registry (Registry) (defaults to: Sourced::Message.registry) —

    resolves type strings to classes. Needs only #all and #[]. Defaults to the root registry, which recurses into every subclass registry, so one codec covers every message type in the process.



117
118
119
120
121
 
# File 'lib/sourced/message/codec.rb', line 117

def initialize(format: self.class.default_format, registry: Sourced::Message.registry)
  @format = format
  @registry = registry
  @messages = nil
end

Instance Attribute Details

#format ⇒ Class<Plumb::Codec> (readonly)

Returns the format compiled onto message types.

Returns:

  • (Class<Plumb::Codec>) —

    the format compiled onto message types



109
110
111
 
# File 'lib/sourced/message/codec.rb', line 109

def format
  @format
end

Class Method Details

.clear_pairs! ⇒ void

This method returns an undefined value.

Discard cached pairs, so every class is compiled again. For a reloader dropping classes, and for a class whose schema changed in place — which reset! cannot detect, since the class is the same object.



104
105
106
 
# File 'lib/sourced/message/codec.rb', line 104

def self.clear_pairs!
  @pairs = nil
end

.default ⇒ Codec

The instance that callers share. Holding no connections or file handles, it is safe to share, so a process compiles its pairs once.

Returns:



65
 
# File 'lib/sourced/message/codec.rb', line 65

def self.default = @default ||= new

.default_format ⇒ Class<Plumb::Codec>

The format a subclass serializes with. Abstract here: Sourced::Message::Codec itself has no format, so it can only be instantiated by passing one explicitly.

Returns:

  • (Class<Plumb::Codec>)

Raises:

  • (NotImplementedError) 


39
40
41
42
43
 
# File 'lib/sourced/message/codec.rb', line 39

def self.default_format
  raise NotImplementedError,
        "#{name} has no format of its own; use a subclass (JSONCodec, FormsCodec) " \
        'or pass format:'
end

.pairs ⇒ Hash{Class => Hash{Class<Plumb::Codec> => Pair}}

Compiled pairs, keyed by message class, shared across every instance of this class and every reset. A class's schema is fixed once its define block has run, so its pair is too — which makes recompiling a matter of re-collecting existing pairs. Redefining a message type produces a new class, so it misses the cache and compiles fresh; that is what makes the cache safe under reloading. (Reopening a class to add attributes after it has been compiled once would not be picked up; clear_pairs! is the way out of that.)

Keyed by the message class, not by the type #compiled_type returns: a class is identity-keyed and stable, where a Plumb node would put the cache at the mercy of that node's +#hash+/+#eql?+.

Held strongly. A weakly-keyed map would not help: a pair is built from its class and refers back to it, so holding the pair keeps the class reachable either way. A reloader that discards classes should call clear_pairs!.

Returns:

  • (Hash{Class => Hash{Class<Plumb::Codec> => Pair}}) 


95
96
97
 
# File 'lib/sourced/message/codec.rb', line 95

def self.pairs
  @pairs ||= {}
end

.reset! ⇒ void

This method returns an undefined value.

Drop the shared instance, so the next default compiles against the current registry — between tests, and for a development-mode class reloader.

The pairs cache deliberately survives: building a pair is the expensive part of a compile, and a class that did not change does not need a new one.



74
75
76
 
# File 'lib/sourced/message/codec.rb', line 74

def self.reset!
  @default = nil
end

Instance Method Details

#compile! ⇒ self

Build a pair for every message class in the registry, frozen once they are all in.

Also the boot check: a message type this format cannot represent raises here, naming the offending attribute path, so the failure lands at boot instead of on the first message that happens to carry the type.

Idempotent, so several collaborators sharing a codec can each call it on start without coordinating. #recompile! is how to pick up types registered since.

Note what it does not check: whether data already written satisfies its schema. That is a per-message question, answered by #decode when the message is read.

Returns:

  • (self)

Raises:

  • (Plumb::TypeError) —

    if any registered message type can't be serialized by this format



147
148
149
150
151
152
153
154
 
# File 'lib/sourced/message/codec.rb', line 147

def compile!
  return self if compiled?

  messages = {}
  @registry.all { |klass| messages[klass.type] = pair_for(klass) }
  @messages = messages.freeze
  self
end

#compiled? ⇒ Boolean

Returns whether #compile! has run.

Returns:



130
 
# File 'lib/sourced/message/codec.rb', line 130

def compiled? = !@messages.nil?

#decode(attrs) ⇒ Sourced::Message

Rebuild a message from decoded attributes. An unknown type raises: a process reading types it doesn't know about is missing the class.

Parameters:

  • attrs (Hash) —

    symbol-keyed message attributes

Returns:

Raises:



190
191
192
193
194
195
196
197
198
 
# File 'lib/sourced/message/codec.rb', line 190

def decode(attrs)
  type = attrs[:type]
  klass = @registry[type]
  raise UnknownMessageError, "Unknown message type: #{label(type, attrs[:id])}" unless klass

  build(klass, attrs, pair(type, attrs[:id]).decoder)
rescue Plumb::ParseError => e
  raise DecodeError, "cannot decode #{label(type, attrs[:id])}: #{e.message}"
end

#encode(message) ⇒ Object

Encode a message into the format's native values, ready to serialize.

Parameters:

Returns:

  • (Object) —

    whatever the format renders — a Hash for JSON

Raises:



176
177
178
179
180
 
# File 'lib/sourced/message/codec.rb', line 176

def encode(message)
  pair(message.type, message.id).encoder.parse(encode_subject(message))
rescue Plumb::ParseError => e
  raise EncodeError, "cannot encode #{label(message.type, message.id)}: #{e.message}"
end

#inspect ⇒ String

attr_reader :format shadows Kernel#format in instance scope, so this interpolates.

Returns:

  • (String) 


127
 
# File 'lib/sourced/message/codec.rb', line 127

def inspect = "#<#{self.class.name} format=#{@format.name}#{compiled? ? '' : ' (not compiled)'}>"

#recompile! ⇒ self

Compile again from scratch, picking up message types and encoders registered since the last compile.

Returns:

  • (self)

Raises:



161
162
163
164
 
# File 'lib/sourced/message/codec.rb', line 161

def recompile!
  @messages = nil
  compile!
end

#registered?(type) ⇒ Boolean

Returns whether a pair was compiled for this type.

Parameters:

  • type (String) —

    message type string

Returns:

  • (Boolean) —

    whether a pair was compiled for this type



168
 
# File 'lib/sourced/message/codec.rb', line 168

def registered?(type) = !@messages.nil? && @messages.key?(type)