alias Maru.Entity.Struct.Batch alias Maru.Entity.Struct.Instance alias Maru.Entity.Struct.Serializer alias Maru.Entity.Struct.Exposure defmodule Maru.Entity.Runtime do @moduledoc """ Serialize an instance concurrently. """ @type id :: reference @type state :: %{ data: :ets.tab(), old_link: :ets.tab(), new_link: :ets.tab(), old_batch: :ets.tab(), new_batch: :ets.tab(), max_concurrency: pos_integer } @doc false @spec serialize(Serializer.t(), Maru.Entity.instance(), Entity.options()) :: Maru.Entity.object() def serialize(serializer, instance, options) do state = init(options) id = save_link(serializer, instance, state.old_link) state = do_loop(state, serializer.options) terminate(id, serializer.type, serializer.options, state) end @spec init(Keyword.t()) :: state defp init(options) do max_concurrency = Keyword.get(options, :max_concurrency) || Application.get_env(:maru_entity, :default_max_concurrency) || System.schedulers_online() * 4 exposure_filter_funcs = for {entity_module, options} <- Keyword.drop(options, [:max_concurrency]) do only = Keyword.get(options, :only, nil) except = Keyword.get(options, :except, nil) attr_groups = flat_attrs(only || except || [], []) filter_func = cond do not is_nil(only) and not is_nil(except) -> raise ":only and :except are in conflict!" not is_nil(only) -> fn exposure -> Enum.any?( attr_groups, &(start_with?(exposure.attr_group, &1) in [true, :exactly]) ) end not is_nil(except) -> fn exposure -> Enum.all?( attr_groups, &(start_with?(exposure.attr_group, &1) in [false, :exactly]) ) end true -> fn _ -> true end end {entity_module, filter_func} end %{ data: create_ets(:duplicate_bag), old_link: create_ets(:duplicate_bag), new_link: create_ets(:duplicate_bag), old_batch: create_ets(:set), new_batch: create_ets(:set), max_concurrency: max_concurrency, exposure_filter_funcs: exposure_filter_funcs } end defp flat_attrs([], result), do: result defp flat_attrs([{key, nested} | t], result) do new_result = for nested_result <- flat_attrs(nested, []) do [key | nested_result] end flat_attrs(t, new_result ++ [[key]] ++ result) end defp flat_attrs([h | t], result) do flat_attrs(t, [[h, :*] | result]) end defp start_with?([], []), do: :exactly defp start_with?(_, [:*]), do: true defp start_with?([h | t1], [h | t2]), do: start_with?(t1, t2) defp start_with?(_, _), do: false @spec terminate(id, Maru.Entity.one_or_many(), Entity.options(), state) :: Maru.Entity.object() defp terminate(id, type, options, state) do :ets.delete(state.old_link) :ets.delete(state.new_link) :ets.delete(state.old_batch) :ets.delete(state.new_batch) data = do_build(id, type, options, state.data) :ets.delete(state.data) data end @spec create_ets(:set | :duplicate_bag) :: :ets.tab() defp create_ets(type) when type in [:set, :duplicate_bag] do :ets.new(:maru_entity_serializer, [ type, :public, heir: :none, write_concurrency: true, read_concurrency: true ]) end @spec save_link(Serializer.t(), Maru.Entity.instance(), :ets.tab()) :: id defp save_link(%Serializer{type: :one} = s, instance, ets) do id = make_ref() :ets.insert(ets, {id, s, instance, nil}) id end defp save_link(%Serializer{type: :many} = s, instances, ets) do id = make_ref() instances |> Stream.with_index() |> Enum.each(fn {i, idx} -> :ets.insert(ets, {id, s, i, idx}) end) id end @spec do_build(id, Maru.Entity.one_or_many(), Entity.options(), :ets.tab()) :: Maru.Entity.object() | [Maru.Entity.object()] defp do_build(id, :one, options, ets) do [{_id, i, nil}] = :ets.lookup(ets, id) do_build_one(i, options, ets) end defp do_build(id, :many, options, ets) do :ets.lookup(ets, id) |> Enum.sort(fn {_, _, idx1}, {_, _, idx2} -> idx1 < idx2 end) |> Enum.map(fn {_id, i, _idx} -> do_build_one(i, options, ets) end) end @spec do_build_one(Instance.t(), Entity.options(), :ets.tab()) :: Maru.Entity.object() defp do_build_one(%Instance{data: data, links: links, module: module}, options, ets) do links |> Enum.reduce(data, fn {attr_group, type, id}, acc -> put_in(acc, attr_group, do_build(id, type, options, ets)) end) |> case do result when is_nil(module) -> result result -> module.before_finish(result, options) end end @spec do_loop(state, Entity.options()) :: state defp do_loop(state, options) do unless :ets.info(state.old_link)[:size] == 0 do do_loop_link_monitor(state) end unless :ets.info(state.old_batch)[:size] == 0 do :ets.delete_all_objects(state.old_link) do_loop_batch(state, options) do_loop_link_monitor(state) end case :ets.info(state.new_link)[:size] + :ets.info(state.new_batch)[:size] do 0 -> state _ -> :ets.delete_all_objects(state.old_link) :ets.delete_all_objects(state.old_batch) %{ state | old_link: state.new_link, new_link: state.old_link, old_batch: state.new_batch, new_batch: state.old_batch } |> do_loop(options) end end @spec do_loop_link_monitor(state) :: :ok defp do_loop_link_monitor(state) do parent = self() {pid, ref} = Process.spawn(fn -> do_loop_link(parent, state) end, [:monitor]) receive do {:DOWN, ^ref, :process, ^pid, :normal} -> :ok {:DOWN, ^ref, :process, ^pid, reason} -> {:error, exception, stack} = reason :erlang.raise(:error, exception, stack) end end @spec do_loop_link(pid, state) :: :ok defp do_loop_link(parent, state) do Process.flag(:trap_exit, true) parent_ref = Process.monitor(parent) sync = fn continue -> receive do {:EXIT, _pid, :normal} -> continue.() {:EXIT, _pid, reason} -> exit(reason) {:DOWN, ^parent_ref, :process, ^parent, _} -> exit(:kill) end end {_, rest} = :ets.foldl( fn term, {0, rest} -> sync.(fn -> Process.spawn(__MODULE__, :do_serialize, [term, state], [:link]) end) {0, rest} term, {num, rest} -> Process.spawn(__MODULE__, :do_serialize, [term, state], [:link]) {num - 1, rest + 1} end, {state.max_concurrency, 0}, state.old_link ) for _ <- :lists.seq(1, rest) do sync.(fn -> :ok end) end :ok end @spec do_loop_batch(state, Entity.options()) :: :ok defp do_loop_batch(state, options) do data = :ets.foldl( fn {id, serializer, %Batch{module: module, key: key}}, acc -> case acc do %{^module => value} -> value = Map.put(value, key, {id, serializer}) Map.put(acc, module, value) _ -> Map.put(acc, module, %{ key => {id, serializer} }) end end, %{}, state.old_batch ) |> Enum.map(fn {module, value} -> {module, Task.async(module, :resolve, [Map.keys(value), options])} end) |> Enum.map(fn {module, task} -> {module, Task.await(task, :infinity)} end) |> Enum.into(%{}) :ets.foldl( fn {id, %Serializer{type: :many} = s, %Batch{module: module, key: key}}, _ -> data |> Map.get(module, %{}) |> Map.get(key, []) |> Stream.with_index() |> Enum.each(fn {instance, idx} -> :ets.insert(state.old_link, {id, %{s | type: :one}, instance, idx}) end) :ok {id, %Serializer{type: :one} = s, %Batch{module: module, key: key}}, _ -> instance = get_in(data, [module, key]) :ets.insert(state.old_link, {id, s, instance, nil}) :ok {id, nil, %Batch{module: module, key: key}}, _ -> instance = get_in(data, [module, key]) :ets.insert(state.old_link, {id, nil, instance, nil}) :ok end, :ok, state.old_batch ) end @doc false @spec do_serialize({id, Serializer.t() | nil, Maru.Entity.instance(), integer}, state) :: :ok def do_serialize({id, nil, instance, idx}, state) do result = %Instance{data: instance, links: []} :ets.insert(state.data, {id, result, idx}) :ok end def do_serialize({id, serializer, instance, idx}, state) do result = case serializer.module.before_serialize(instance, serializer.options) do {:halt, result} -> %Instance{module: serializer.module, data: result} {:ok, instance, options, data} -> exposure_filter_func = Keyword.get( state.exposure_filter_funcs, serializer.module, fn _ -> true end ) exposures = Enum.filter(serializer.module.__exposures__(), exposure_filter_func) do_serialize( exposures, %Instance{module: serializer.module, data: data}, instance, options, state ) {:ok, attrs, instance, options, data} -> exposure_filter_func = Keyword.get( state.exposure_filter_funcs, serializer.module, fn _ -> true end ) exposures = serializer.module.__exposures__() |> Enum.filter(exposure_filter_func) |> Enum.filter(fn exposure -> exposure.attr_group in Enum.map(attrs, &List.wrap/1) end) do_serialize( exposures, %Instance{module: serializer.module, data: data}, instance, options, state ) end :ets.insert(state.data, {id, result, idx}) :ok rescue e -> exit({:error, e, __STACKTRACE__}) end @spec do_serialize( list(Exposure.Runtime.t()), Maru.Entity.object(), Maru.Entity.instance(), Maru.Entity.options(), state ) :: Maru.Entity.object() defp do_serialize([], result, _instance, _options, _state), do: result defp do_serialize([h | t], result, instance, options, state) do h.if_func.(instance, options) |> case do true -> try do value = get_in(result.data, h.attr_group) || case h.do_func do f when is_function(f, 1) -> f.(instance) f when is_function(f, 2) -> f.(instance, options) f when is_function(f, 3) -> f.(instance, options, result.data) end {:ok, (is_nil(value) && h.default) || value} rescue e -> result.module.handle_error(h.attr_group, e, result.data) end false -> :skip end |> case do {:ok, value} -> result = do_update(h.serializer, value, h, result, options, state) do_serialize(t, result, instance, options, state) {:halt, data} -> %Instance{result | data: data, links: []} :skip -> do_serialize(t, result, instance, options, state) end end @spec do_update( Serializer.t() | nil, Batch.t() | Maru.Entity.instance(), Exposure.Runtime.t(), Maru.Entity.instance(), Maru.Entity.options(), state ) :: Maru.Entity.instance() defp do_update(nil, %Batch{} = batch, field, result, _options, state) do id = make_ref() :ets.insert(state.new_batch, {id, nil, batch}) %Instance{ result | data: put_in(result.data, field.attr_group, :unlinked), links: [{field.attr_group, :one, id} | result.links] } end defp do_update(serializer, %Batch{} = batch, field, result, options, state) do id = make_ref() s = %{serializer | options: options} :ets.insert(state.new_batch, {id, s, batch}) %Instance{ result | data: put_in(result.data, field.attr_group, :unlinked), links: [{field.attr_group, serializer.type, id} | result.links] } end defp do_update(nil, instance, field, result, _options, _state) do %{result | data: put_in(result.data, field.attr_group, instance)} end defp do_update(serializer, instance, field, result, options, state) do s = %{serializer | options: options} id = save_link(s, instance, state.new_link) %Instance{ result | data: put_in(result.data, field.attr_group, :unlinked), links: [{field.attr_group, serializer.type, id} | result.links] } end end