Kinesis Writer

writer.ex

defmodule KinesisWriter.Writer do
  @type t :: %__MODULE__{
          aggregator: ExKpl.t(),
          client: module(),
          client_opts: Keyword.t()
        }

  defstruct aggregator: ExKpl.new(),
            client: KinesisWriter.Client,
            client_opts: []

  def new(opts \\ []) do
    struct(__MODULE__, opts)
  end

  def write(%__MODULE__{aggregator: agg} = state, message) do
    case ExKpl.add(agg, {partition_key(), message}) do
      {:undefined, updated_agg} ->
        %{state | aggregator: updated_agg}

      {{pk, full_record, _ehk}, updated_agg} ->
        state.client.put_record(format_record(pk, full_record), state.client_opts)
        %{state | aggregator: updated_agg}
    end
  end

  def flush(%{aggregator: agg, client: client} = state) do
    {{pk, record, _}, new_agg} = ExKpl.finish(agg)
    client.put_record(format_record(pk, record), state.client_opts)
    %{state | aggregator: new_agg}
  end
  
  defp format_record(pk, record), do: %{data: record, partition_key: pk}

  defp partition_key, do: Ecto.UUID.generate()
end