Packages
avrora
0.1.0-beta
0.30.2
0.30.1
0.30.0
0.29.2
0.29.1
0.29.0
0.28.0
0.27.0
0.26.0
0.25.0
0.24.2
0.24.1
0.24.0
0.23.0
0.22.0
0.21.1
0.21.0
0.20.0
0.19.0
0.18.1
0.18.0
0.17.0
0.16.0
0.15.0
0.14.1
0.14.0
0.13.0
0.12.0
0.11.0
0.10.0
0.9.1
0.9.0
0.8.1
0.8.0
0.7.1
0.7.0
0.6.0
0.5.1
0.5.0
0.4.0
0.3.1
0.3.0
0.2.2
0.2.1
0.2.0
0.1.1-beta
0.1.0-beta
An Elixir library for working with Avro messages conveniently. It supports local schema files and Confluent® schema registry.
Current section
Files
Jump to
Current section
Files
lib/aurora/encoder.ex
defmodule Avrora.Encoder do
@moduledoc """
Encodes and decodes avro messages created with or without extra Schema Registry
version.
"""
require Logger
alias Avrora.{Name, Resolver}
@registry_magic_byte <<0::size(8)>>
@doc """
Decodes given message with a schema eather loaded from the local file or from
the configured schema registry.
## Examples
...> payload = <<72, 48, 48, 48, 48, 48, 48, 48, 48, 45, 48, 48, 48, 48, 45,
48, 48, 48, 48, 45, 48, 48, 48, 48, 45, 48, 48, 48, 48, 48, 48, 48, 48, 48,
48, 48, 48, 123, 20, 174, 71, 225, 250, 47, 64>>
...> Avrora.Encoder.decode(payload, schema_name: "io.confluent.Payment")
{:ok, %{"id" => "00000000-0000-0000-0000-000000000000", "amount" => 15.99}}
"""
@spec decode(binary(), keyword(String.t())) :: {:ok, map()} | {:error, term()}
def decode(payload, schema_name: schema_name) when is_binary(payload) do
with {:ok, schema_name} <- Name.parse(schema_name) do
unless is_nil(schema_name.version) do
Logger.warn(
"decoding message with schema version is not allowed, `#{schema_name.name}` used instead"
)
end
{schema_name, body} =
case payload do
<<@registry_magic_byte, <<version::size(32)>>, body::binary>> ->
{"#{schema_name.name}:#{version}", body}
<<body::binary>> ->
{schema_name.name, body}
end
with {:ok, avro} <- Resolver.resolve(schema_name), do: AvroEx.decode(avro.ex_schema, body)
end
end
@doc """
Encodes given message with a schema eather loaded from the local file or from
the configured schema registry.
## Examples
...> payload = %{"id" => "00000000-0000-0000-0000-000000000000", "amount" => 15.99}
...> Avrora.Encoder.encode(payload, schema_name: "io.confluent.Payment")
{:ok, <<72, 48, 48, 48, 48, 48, 48, 48, 48, 45, 48, 48, 48, 48, 45,
48, 48, 48, 48, 45, 48, 48, 48, 48, 45, 48, 48, 48, 48, 48, 48, 48, 48, 48,
48, 48, 48, 123, 20, 174, 71, 225, 250, 47, 64>>}
"""
@spec encode(map(), keyword(String.t())) :: {:ok, binary()} | {:error, term()}
def encode(payload, schema_name: schema_name) when is_map(payload) do
with {:ok, schema_name} <- Name.parse(schema_name),
{:ok, avro} <- Resolver.resolve(schema_name.name),
{:ok, body} <- AvroEx.encode(avro.ex_schema, payload) do
unless is_nil(schema_name.version) do
Logger.warn(
"encoding message with schema version is not allowed, `#{schema_name.name}` used instead"
)
end
body =
if is_nil(avro.version),
do: body,
else: <<@registry_magic_byte, <<avro.version::size(32)>>, body::binary>>
{:ok, body}
end
end
end