Current section

Files

Jump to
avrora lib avrora schema encoder.ex
Raw

lib/avrora/schema/encoder.ex

defmodule Avrora.Schema.Encoder do
@moduledoc """
Encodes and parses Avro schemas from various formats, like JSON and erlavro into `Avrora.Schema`.
"""
alias Avrora.Config
alias Avrora.Schema
alias Avrora.Schema.ReferenceCollector
@type reference_lookup_fun :: (String.t() -> {:ok, String.t()} | {:error, term()})
@reference_lookup_fun &__MODULE__.reference_lookup/1
@doc """
Parse Avro schema JSON and convert to the Schema struct.
## Examples
iex> json = ~s({"namespace":"io.confluent","type":"record","name":"Payment","fields":[{"name":"id","type":"string"},{"name":"amount","type":"double"}]})
iex> {:ok, schema} = Avrora.Schema.Encoder.from_json(json)
iex> schema.full_name
"io.confluent.Payment"
"""
@spec from_json(String.t(), reference_lookup_fun) :: {:ok, Schema.t()} | {:error, term()}
def from_json(payload, reference_lookup_fun \\ @reference_lookup_fun) when is_binary(payload) do
lookup_table = ets().new()
with {:ok, [schema | _]} <- parse_recursive(payload, lookup_table, reference_lookup_fun),
{:ok, full_name} <- extract_full_name(schema),
{:ok, schema} <- do_compile(full_name, lookup_table) do
{
:ok,
%Schema{
id: nil,
version: nil,
full_name: full_name,
lookup_table: lookup_table,
json: to_json(schema)
}
}
else
{:error, reason} ->
true = :ets.delete(lookup_table)
{:error, reason}
end
end
@doc """
An example of a reference lookup which returns empty JSON body
"""
@spec reference_lookup(String.t()) :: {:ok, String.t()} | {:error, term()}
def reference_lookup(_), do: {:ok, ~s({})}
@doc """
Convert `erlavro` format to the Schema struct.
## Examples
iex> payload =
...> {:avro_record_type, "Payment", "io.confluent", "", [],
...> [
...> {:avro_record_field, "id", "", {:avro_primitive_type, "string", []}, :undefined,
...> :ascending, []},
...> {:avro_record_field, "amount", "", {:avro_primitive_type, "double", []}, :undefined,
...> :ascending, []}
...> ], "io.confluent.Payment", []}
iex> {:ok, schema} = Avrora.Schema.Encoder.from_erlavro(payload)
iex> schema.id
nil
iex> schema.full_name
"io.confluent.Payment"
"""
@spec from_erlavro(term(), keyword()) :: {:ok, Schema.t()} | {:error, term()}
def from_erlavro(schema, attributes \\ []) do
lookup_table = ets().new()
with {:ok, full_name} <- extract_full_name(schema),
lookup_table <- :avro_schema_store.add_type(schema, lookup_table),
json <- Keyword.get_lazy(attributes, :json, fn -> to_json(schema) end) do
{
:ok,
%Schema{
id: nil,
version: nil,
full_name: full_name,
lookup_table: lookup_table,
json: json
}
}
else
{:error, reason} ->
true = :ets.delete(lookup_table)
{:error, reason}
end
end
@doc """
Convert struct to `erlavro` format and look it up in `avro_schema_store`.
## Examples
iex> json = ~s({"namespace":"io.confluent","type":"record","name":"Payment","fields":[{"name":"id","type":"string"},{"name":"amount","type":"double"}]})
iex> {:ok, schema} = Avrora.Schema.Encoder.from_json(json)
iex> {:ok, {type, _, _, _, _, _, full_name, _}} = Avrora.Schema.Encoder.to_erlavro(schema)
iex> full_name
"io.confluent.Payment"
iex> type
:avro_record_type
"""
@spec to_erlavro(Schema.t()) :: {:ok, term()} | {:error, term()}
def to_erlavro(%Schema{} = schema),
do: do_compile(schema.full_name, schema.lookup_table)
defp to_json(schema), do: :avro_json_encoder.encode_type(schema)
defp parse_recursive(payload, lookup_table, reference_lookup_fun) do
with {:ok, schema} <- do_parse(payload),
{:ok, _} <- extract_full_name(schema),
{:ok, references} <- ReferenceCollector.collect(schema),
lookup_table <- :avro_schema_store.add_type(schema, lookup_table) do
payloads =
references
|> Enum.reject(&:avro_schema_store.lookup_type(&1, lookup_table))
|> Enum.map(fn reference ->
reference |> reference_lookup_fun.() |> unwrap!()
end)
schemas =
Enum.flat_map(payloads, fn payload ->
payload |> parse_recursive(lookup_table, reference_lookup_fun) |> unwrap!()
end)
{:ok, [schema | schemas]}
end
catch
error -> {:error, error}
end
defp unwrap!({:ok, result}), do: result
defp unwrap!({:error, error}), do: throw(error)
defp extract_full_name(schema) do
case schema do
{:avro_fixed_type, _, _, _, _, full_name, _} -> {:ok, full_name}
{:avro_enum_type, _, _, _, _, _, full_name, _} -> {:ok, full_name}
{:avro_record_type, _, _, _, _, _, full_name, _} -> {:ok, full_name}
_ -> {:error, :unnamed_type}
end
end
# Compile complete version of the `erlavro` format with all references
# being resolved, converting errors to error return
defp do_compile(full_name, lookup_table) do
{:ok, :avro_util.expand_type(full_name, lookup_table)}
rescue
_ in MatchError -> {:error, :bad_reference}
error in ErlangError -> {:error, error.original}
end
# Parse schema to `erlavro` format, converting errors to error return
defp do_parse(payload) do
{:ok, :avro_json_decoder.decode_schema(payload, allow_bad_references: true)}
rescue
error in ArgumentError -> {:error, error.message}
error in ErlangError -> {:error, error.original}
end
defp ets, do: Config.self().ets_lib()
end