defmodule Grains.GenGrain do use GenServer @bread_key :bread @successors_key :successors @predecessors_key :predecessors @process_map_key :process_map @own_name_key :own_name require Logger # @callback handle_pull defmodule State do @moduledoc false @enforce_keys [:substate, :mod] defstruct @enforce_keys end defmacro __using__(_args) do quote do use GenServer defdelegate push(short_name, msg), to: Grains.GenGrain defdelegate push(msg), to: Grains.GenGrain defdelegate pull(from), to: Grains.GenGrain defdelegate pull_with_tag(from, tag), to: Grains.GenGrain defdelegate pull(), to: Grains.GenGrain defdelegate pull_with_tag(tag), to: Grains.GenGrain defdelegate own_name(), to: Grains.GenGrain defdelegate own_full_name(), to: Grains.GenGrain end end def start_link(mod, bread, name, args, opts) do GenServer.start_link(__MODULE__, {mod, bread, name, args}, opts) end def init({mod, bread, short_name, args}) do store(short_name, bread) case mod.init(args) do {:ok, substate} -> {:ok, %State{substate: substate, mod: mod}} {:ok, substate, ret} -> {:ok, %State{substate: substate, mod: mod}, ret} other -> other end end def handle_call({:debug, :reply_chain, sender, reply_chain, message}, _from, state) do {:reply, debug_reply_chain_forward(reply_chain, sender, message), state} end def handle_call(msg, from, state = %State{mod: mod, substate: sub}) do mod.handle_call(msg, from, sub) |> handle_return(state) end def handle_cast(msg, state = %State{mod: mod, substate: sub}) do mod.handle_cast(msg, sub) |> handle_return(state) end def handle_info({:debug, :reply_chain, sender, reply_chain, message}, state) do debug_reply_chain_forward(reply_chain, sender, message) {:noreply, state} end def handle_info({:push, short_from, msg}, state = %State{mod: mod, substate: sub}) do case mod.handle_push(msg, short_from, sub) do {:noreply, sub} -> {:noreply, %State{state | substate: sub}} {:noreply, sub, ret} -> {:noreply, %State{state | substate: sub}, ret} end end def handle_info({:pull, short_from}, state = %State{mod: mod, substate: sub}) do mod.handle_pull(short_from, sub) |> handle_return(state) |> handle_pull_return(short_from) end def handle_info({:pull, tag, short_from}, state = %State{mod: mod, substate: sub}) do mod.handle_pull(short_from, tag, sub) |> handle_return(state) |> handle_pull_return(short_from) end def handle_info(msg, state = %State{mod: mod, substate: sub}) do mod.handle_info(msg, sub) |> handle_return(state) end defp debug_reply_chain_forward([last], sender, message) do case debug_reply_chain_check_current([last]) do :ok -> send(sender, message) e = {:error, _} -> Logger.error("Debug reply chain: #{inspect(e)}") e end end defp debug_reply_chain_forward(reply_chain = [_current, next | rest], sender, message) do with :ok <- debug_reply_chain_check_current(reply_chain), :ok <- send_with_error(next, {:debug, :reply_chain, sender, [next | rest], message}) do :ok else e = {:error, _} -> Logger.error("Debug reply chain: #{inspect(e)}") end end defp debug_reply_chain_check_current([current | _]) do own_name = own_name() own_full_name = own_full_name() if current in [own_name, own_full_name] do :ok else {:error, [reply_chain_current: current, expected_one_of: [own_name, own_full_name]]} end end def handle_continue(term, state = %State{mod: mod, substate: sub}) do mod.handle_continue(term, sub) |> handle_return(state) end defp handle_return(return, state) do case return do {:noreply, sub} -> {:noreply, %State{state | substate: sub}} {:noreply, sub, ret} -> {:noreply, %State{state | substate: sub}, ret} {:reply, reply, sub} -> {:reply, reply, %State{state | substate: sub}} {:reply, reply, sub, ret} -> {:reply, reply, %State{state | substate: sub}, ret} end end defp handle_pull_return(return, short_from) do case return do {:reply, reply, state} -> push(short_from, reply) {:noreply, state} {:reply, reply, state, ret} -> push(short_from, reply) {:noreply, state, ret} other -> other end end def pull(pred) do if Enum.member?(predecessors(), pred) do send_pull(pred, own_name()) :ok else {:error, {:push_to_unknown, pred}} end end def pull() do case predecessors() do [] -> {:error, :no_predecessors} preds -> name = own_name() Enum.map(preds, fn s -> send_pull(s, name) end) :ok end end def pull_with_tag(pred, tag) do if Enum.member?(predecessors(), pred) do send_pull(pred, own_name(), tag) :ok else {:error, {:push_to_unknown, pred}} end end def pull_with_tag(tag) do case predecessors() do [] -> {:error, :no_predecessors} preds -> name = own_name() Enum.map(preds, fn s -> send_pull(s, name, tag) end) :ok end end def push(short_name, msg) do if Enum.member?(successors(), short_name) do send_pub(short_name, msg) :ok else {:error, {:push_to_unknown, short_name}} end end def push(msg) do case successors() do [] -> {:error, :no_successors} succs -> Enum.map(succs, fn s -> send_pub(s, msg) end) :ok end end defp send_pub(grain, msg) do maybe_send(grain, {:push, own_name(), msg}) end defp send_pull(grain, name) do maybe_send(grain, {:pull, name}) end defp send_pull(grain, name, tag) do maybe_send(grain, {:pull, tag, name}) end defp maybe_send(grain, msg) do case Process.whereis(full_name(grain)) do nil -> :ok pid -> send(pid, msg) end end defp send_with_error(grain, msg) do case Process.whereis(full_name(grain)) do nil -> {:error, {:unknown_grain, grain}} pid -> send(pid, msg) :ok end end defp store(short_name, bread) do Process.put(@bread_key, bread) Process.put(@process_map_key, bread.process_map) Process.put(@own_name_key, short_name) Enum.filter(bread.final_recipe, fn {l, _} -> l == short_name end) |> Enum.map(fn {_, r} -> r end) |> (&Process.put(@successors_key, &1)).() Enum.filter(bread.final_recipe, fn {_, r} -> r == short_name end) |> Enum.map(fn {l, _} -> l end) |> (&Process.put(@predecessors_key, &1)).() end def full_name(short_name) do process_map() |> Map.fetch!(short_name) end def own_name() do Process.get(@own_name_key) end def own_full_name do own_name() |> full_name() end defp process_map() do Process.get(@process_map_key) end defp successors() do Process.get(@successors_key) end defp predecessors() do Process.get(@predecessors_key) end end