adds decode/3 with format option

lib/avrora.ex

@@ -6,6 +6,8 @@ defmodule Avrora do
   defdelegate decode(payload), to: Avrora.Encoder
   defdelegate encode(payload, opts), to: Avrora.Encoder
   defdelegate decode(payload, opts), to: Avrora.Encoder
+  defdelegate decode_plain(payload, opts), to: Avrora.Encoder
+  defdelegate encode_plain(payload, opts), to: Avrora.Encoder
   defdelegate extract_schema(payload), to: Avrora.Encoder
 
   def start_link(opts \\ []), do: Supervisor.start_link(__MODULE__, opts, name: __MODULE__)

lib/avrora/encoder.ex

@@ -51,6 +51,29 @@ defmodule Avrora.Encoder do
     codec.decode(payload)
   end
 
+  @doc """
+  Decode binary Avro message with :plain format and given schema.
+
+  ## Examples
+
+      ...> payload = <<0, 232, 220, 144, 233, 11, 200, 1>>
+      ...> Avrora.Encoder.decode_plain(payload,"io.confluent.NumericTransfer")
+      {:ok, %{ "link_is_enabled" => false, "updated_at" => 1_586_632_500, "updated_by_id" => 1_00 }
+  """
+  @spec decode_plain(binary(), schema_name: String.t()) ::
+          {:ok, map() | list(map())} | {:error, term()}
+  def decode_plain(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 supported, `#{schema_name.name}` used instead"
+        )
+      end
+
+      Codec.Plain.decode(payload, schema: %Schema{full_name: schema_name.name})
+    end
+  end
+
   @doc """
   Decode binary Avro message, loading schema from local file or Schema Registry.
 
@@ -136,4 +159,25 @@ defmodule Avrora.Encoder do
       end
     end
   end
+
+  @doc """
+  Encode binary Avro message with :plain format and given schema.
+
+  ## Examples
+
+      ...> payload = %{ "link_is_enabled" => false, "updated_at" => 1_586_632_500, "updated_by_id" => 1_00 }
+      ...> Avrora.Encoder.encode_plain(payload,"io.confluent.NumericTransfer")
+      {:ok, <<0, 232, 220, 144, 233, 11, 200, 1>>}
+  """
+  def encode_plain(payload, schema_name: schema_name) when is_map(payload) do
+    with {:ok, schema_name} <- Name.parse(schema_name) do
+      unless is_nil(schema_name.version) do
+        Logger.warn(
+          "encoding message with schema version is not supported yet, `#{schema_name.name}` used instead"
+        )
+      end
+
+      Codec.Plain.encode(payload, schema: %Schema{full_name: schema_name.name})
+    end
+  end
 end

test/avrora/encoder_test.exs

@@ -402,6 +402,53 @@ defmodule Avrora.EncoderTest do
     end
   end
 
+  describe "decode_plain/2" do
+    test "when decoding plain message that starts with what looks like a magic byte" do
+      schema_name = "io.confluent.NumericTransfer"
+      numeric_transfer_schema = numeric_transfer_schema()
+
+      Avrora.Storage.MemoryMock
+      |> expect(:get, fn key ->
+        assert key == "io.confluent.NumericTransfer"
+
+        {:ok, nil}
+      end)
+      |> expect(:put, fn key, value ->
+        assert key == "io.confluent.NumericTransfer"
+        assert value == numeric_transfer_schema
+
+        {:ok, value}
+      end)
+
+      Avrora.Storage.RegistryMock
+      |> expect(:put, fn key, value ->
+        assert key == "io.confluent.NumericTransfer"
+        assert value == numeric_transfer_json_schema()
+
+        {:error, :unconfigured_registry_url}
+      end)
+
+      Avrora.Storage.FileMock
+      |> expect(:get, fn key ->
+        assert key == "io.confluent.NumericTransfer"
+
+        {:ok, numeric_transfer_schema}
+      end)
+
+      {:ok, decoded} =
+        Avrora.decode_plain(
+          numeric_transfer_plain_message_with_fake_magic_byte(),
+          schema_name: schema_name
+        )
+
+      assert decoded == %{
+               "link_is_enabled" => false,
+               "updated_at" => 1_586_632_500,
+               "updated_by_id" => 1_00
+             }
+    end
+  end
+
   describe "encode/2" do
     test "when registry is not configured" do
       payment_schema = payment_schema()
@@ -706,6 +753,10 @@ defmodule Avrora.EncoderTest do
       119, 32, 97, 114, 101, 32, 121, 111, 117, 63, 0>>
   end
 
+  defp numeric_transfer_plain_message_with_fake_magic_byte do
+    <<0, 232, 220, 144, 233, 11, 200, 1>>
+  end
+
   defp payment_schema do
     {:ok, schema} = Schema.parse(payment_json_schema())
     %{schema | id: nil, version: nil}
@@ -721,6 +772,11 @@ defmodule Avrora.EncoderTest do
     %{schema | id: nil, version: nil}
   end
 
+  defp numeric_transfer_schema do
+    {:ok, schema} = Schema.parse(numeric_transfer_json_schema())
+    %{schema | id: nil, version: nil}
+  end
+
   defp messenger_json_schema do
     ~s({"namespace":"io.confluent","name":"Messenger","type":"record","fields":[{"name":"inbox","type":{"type":"array","items":{"type":"record","name":"Message","fields":[{"name":"text","type":"string"}]}}},{"name":"archive","type":{"type":"array","items":"io.confluent.Message"}}]})
   end
@@ -732,4 +788,8 @@ defmodule Avrora.EncoderTest do
   defp payment_json_schema do
     ~s({"namespace":"io.confluent","name":"Payment","type":"record","fields":[{"name":"id","type":"string"},{"name":"amount","type":"double"}]})
   end
+
+  defp numeric_transfer_json_schema do
+    ~s({"namespace":"io.confluent","name":"NumericTransfer","type":"record","fields":[{"name":"link_is_enabled","type":"boolean"},{"name":"updated_at","type":"int"},{"name":"updated_by_id","type":"int"}]})
+  end
 end

test/integration/lib/interfaces.ex

@@ -34,6 +34,11 @@ defmodule Interfaces do
       Avrora.decode(<<2, 118>>, schema_name: "avrora.Record")
     end
 
+    @doc false
+    def decode_plain_with_one_option do
+      Avrora.decode(<<2, 118>>, schema_name: "avrora.Record")
+    end
+
     @doc false
     def encode_with_one_option do
       Avrora.encode(%{"k" => "v"}, schema_name: "avrora.Record")
@@ -43,5 +48,10 @@ defmodule Interfaces do
     def encode_with_two_options do
       Avrora.encode(%{"k" => "v"}, schema_name: "avrora.Record", format: :plain)
     end
+
+    # doc false
+    def encode_plain_with_one_option do
+      Avrora.encode_plain(%{"k" => "v"}, schema_name: "avrora.Record")
+    end
   end
 end