Class: AvroTurf::Messaging
- Inherits:
-
Object
- Object
- AvroTurf::Messaging
- Defined in:
- lib/avro_turf/messaging.rb
Overview
Provides a way to encode and decode messages without having to embed schemas in the encoded data. Confluent’s Schema Registry is used to register a schema when encoding a message – the registry will issue a schema id that will be included in the encoded data alongside the actual message. When decoding the data, the schema id will be used to look up the writer’s schema from the registry.
Defined Under Namespace
Classes: DecodedMessage
Constant Summary collapse
- MAGIC_BYTE =
[0].pack("C").freeze
Instance Method Summary collapse
-
#decode(data, schema_name: nil, namespace: @namespace) ⇒ Object
Decodes data into the original message.
-
#decode_message(data, schema_name: nil, namespace: @namespace) ⇒ Object
Decodes data into the original message.
-
#encode(message, schema_name: nil, namespace: @namespace, subject: nil, version: nil, schema_id: nil, validate: false) ⇒ Object
Encodes a message using the specified schema.
-
#fetch_schema(subject:, version: 'latest') ⇒ Object
Providing subject and version to determine the schema, which skips the auto registeration of schema on the schema registry.
-
#fetch_schema_by_id(schema_id) ⇒ Object
Fetch the schema from registry with the provided schema_id.
-
#initialize(registry: nil, registry_url: nil, schema_store: nil, schemas_path: nil, namespace: nil, registry_path_prefix: nil, logger: nil, proxy: nil, user: nil, password: nil, ssl_ca_file: nil, client_cert: nil, client_key: nil, client_key_pass: nil, client_cert_data: nil, client_key_data: nil) ⇒ Messaging
constructor
Instantiate a new Messaging instance with the given configuration.
-
#register_schema(schema_name:, subject: nil, namespace: nil) ⇒ Object
Schemas are registered under the full name of the top level Avro record type, or ‘subject` if it’s provided.
Constructor Details
#initialize(registry: nil, registry_url: nil, schema_store: nil, schemas_path: nil, namespace: nil, registry_path_prefix: nil, logger: nil, proxy: nil, user: nil, password: nil, ssl_ca_file: nil, client_cert: nil, client_key: nil, client_key_pass: nil, client_cert_data: nil, client_key_data: nil) ⇒ Messaging
Instantiate a new Messaging instance with the given configuration.
registry - A schema registry object that responds to all methods in the
AvroTurf::ConfluentSchemaRegistry interface.
registry_url - The String URL of the schema registry that should be used. schema_store - A schema store object that responds to #find(schema_name, namespace). schemas_path - The String file system path where local schemas are stored. namespace - The String default schema namespace. registry_path_prefix - The String URL path prefix used to namespace schema registry requests (optional). logger - The Logger that should be used to log information (optional). proxy - Forward the request via proxy (optional). user - User for basic auth (optional). password - Password for basic auth (optional). ssl_ca_file - Name of file containing CA certificate (optional). client_cert - Name of file containing client certificate (optional). client_key - Name of file containing client private key to go with client_cert (optional). client_key_pass - Password to go with client_key (optional). client_cert_data - In-memory client certificate (optional). client_key_data - In-memory client private key to go with client_cert_data (optional).
58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 |
# File 'lib/avro_turf/messaging.rb', line 58 def initialize( registry: nil, registry_url: nil, schema_store: nil, schemas_path: nil, namespace: nil, registry_path_prefix: nil, logger: nil, proxy: nil, user: nil, password: nil, ssl_ca_file: nil, client_cert: nil, client_key: nil, client_key_pass: nil, client_cert_data: nil, client_key_data: nil ) @logger = logger || Logger.new($stderr) @namespace = namespace @schema_store = schema_store || SchemaStore.new(path: schemas_path || DEFAULT_SCHEMAS_PATH) @registry = registry || CachedConfluentSchemaRegistry.new( ConfluentSchemaRegistry.new( registry_url, logger: @logger, proxy: proxy, user: user, password: password, ssl_ca_file: ssl_ca_file, client_cert: client_cert, client_key: client_key, client_key_pass: client_key_pass, client_cert_data: client_cert_data, client_key_data: client_key_data, path_prefix: registry_path_prefix ) ) @schemas_by_id = {} end |
Instance Method Details
#decode(data, schema_name: nil, namespace: @namespace) ⇒ Object
Decodes data into the original message.
data - A String containing encoded data. schema_name - The String name of the schema that should be used to decode
the data. Must match the schema used when encoding (optional).
namespace - The namespace of the schema (optional).
Returns the decoded message.
161 162 163 |
# File 'lib/avro_turf/messaging.rb', line 161 def decode(data, schema_name: nil, namespace: @namespace) (data, schema_name: schema_name, namespace: namespace). end |
#decode_message(data, schema_name: nil, namespace: @namespace) ⇒ Object
Decodes data into the original message.
data - A String containing encoded data. schema_name - The String name of the schema that should be used to decode
the data. Must match the schema used when encoding (optional).
namespace - The namespace of the schema (optional).
Returns Struct with the next attributes:
schema_id - The integer id of schema used to encode the message
message - The decoded message
175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 |
# File 'lib/avro_turf/messaging.rb', line 175 def (data, schema_name: nil, namespace: @namespace) readers_schema = schema_name && @schema_store.find(schema_name, namespace) stream = StringIO.new(data) decoder = Avro::IO::BinaryDecoder.new(stream) # The first byte is MAGIC!!! magic_byte = decoder.read(1) if magic_byte != MAGIC_BYTE raise "Expected data to begin with a magic byte, got `#{magic_byte.inspect}`" end # The schema id is a 4-byte big-endian integer. schema_id = decoder.read(4).unpack("N").first writers_schema = @schemas_by_id.fetch(schema_id) do schema_json = @registry.fetch(schema_id) @schemas_by_id[schema_id] = Avro::Schema.parse(schema_json) end reader = Avro::IO::DatumReader.new(writers_schema, readers_schema) = reader.read(decoder) DecodedMessage.new(schema_id, writers_schema, readers_schema, ) rescue Excon::Error::NotFound raise SchemaNotFoundError.new("Schema with id: #{schema_id} is not found on registry") end |
#encode(message, schema_name: nil, namespace: @namespace, subject: nil, version: nil, schema_id: nil, validate: false) ⇒ Object
Encodes a message using the specified schema.
message - The message that should be encoded. Must be compatible with
the schema.
schema_name - The String name of the schema that should be used to encode
the data.
namespace - The namespace of the schema (optional). subject - The subject name the schema should be registered under in
the schema registry (optional).
version - The integer version of the schema that should be used to decode
the data. Must match the schema used when encoding (optional).
schema_id - The integer id of the schema that should be used to encode
the data.
validate - The boolean for performing complete message validation before
encoding it, Avro::SchemaValidator::ValidationError with
a descriptive message will be raised in case of invalid message.
Returns the encoded data as a String.
116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 |
# File 'lib/avro_turf/messaging.rb', line 116 def encode(, schema_name: nil, namespace: @namespace, subject: nil, version: nil, schema_id: nil, validate: false) schema, schema_id = if schema_id fetch_schema_by_id(schema_id) elsif subject && version fetch_schema(subject: subject, version: version) elsif schema_name register_schema(subject: subject, schema_name: schema_name, namespace: namespace) else raise ArgumentError.new('Neither schema_name nor schema_id nor subject + version provided to determine the schema.') end if validate Avro::SchemaValidator.validate!(schema, , recursive: true, encoded: false, fail_on_extra_fields: true) end stream = StringIO.new writer = Avro::IO::DatumWriter.new(schema) encoder = Avro::IO::BinaryEncoder.new(stream) # Always start with the magic byte. encoder.write(MAGIC_BYTE) # The schema id is encoded as a 4-byte big-endian integer. encoder.write([schema_id].pack("N")) # The actual message comes last. writer.write(, encoder) stream.string rescue Excon::Error::NotFound if schema_id raise SchemaNotFoundError.new("Schema with id: #{schema_id} is not found on registry") else raise SchemaNotFoundError.new("Schema with subject: `#{subject}` version: `#{version}` is not found on registry") end end |
#fetch_schema(subject:, version: 'latest') ⇒ Object
Providing subject and version to determine the schema, which skips the auto registeration of schema on the schema registry. Fetch the schema from registry with the provided subject name and version.
206 207 208 209 210 211 |
# File 'lib/avro_turf/messaging.rb', line 206 def fetch_schema(subject:, version: 'latest') schema_data = @registry.subject_version(subject, version) schema_id = schema_data.fetch('id') schema = Avro::Schema.parse(schema_data.fetch('schema')) [schema, schema_id] end |
#fetch_schema_by_id(schema_id) ⇒ Object
Fetch the schema from registry with the provided schema_id.
214 215 216 217 218 219 220 |
# File 'lib/avro_turf/messaging.rb', line 214 def fetch_schema_by_id(schema_id) schema = @schemas_by_id.fetch(schema_id) do schema_json = @registry.fetch(schema_id) Avro::Schema.parse(schema_json) end [schema, schema_id] end |
#register_schema(schema_name:, subject: nil, namespace: nil) ⇒ Object
Schemas are registered under the full name of the top level Avro record type, or ‘subject` if it’s provided.
224 225 226 227 228 |
# File 'lib/avro_turf/messaging.rb', line 224 def register_schema(schema_name:, subject: nil, namespace: nil) schema = @schema_store.find(schema_name, namespace) schema_id = @registry.register(subject || schema.fullname, schema) [schema, schema_id] end |