defmodule Grains do @moduledoc """ Grains describes data flow as a graph with interchangeable parts. To accomplish that grains divides the graph (`Recipe`) from the implementation of the nodes (`Grains`). Together these parts can be used to make a `Bread` which describes how the graph translates to processes. Each process has a symbolic short name, which is resolved to the registered name of the process via a map (`%Bread{}.process_map`). """ alias Grains.Bread alias Grains.Recipe alias Grains.Supervisor defstruct [:map] def new(map) do %__MODULE__{map: map} end @doc """ Merges two grain maps into one. The function works the same as `Map.merge/2`: all grains in the second argument are added to the first, overriding any existing one. """ def merge(%__MODULE__{map: a}, %__MODULE__{map: b}) do a |> Map.merge(b) |> new() end defdelegate new_recipe(name, new), to: Grains.Recipe, as: :new defdelegate merge_recipes(name, a, b), to: Grains.Recipe, as: :merge defdelegate start_supervised(recipe, grains, args \\ []), to: Grains.Supervisor, as: :start_link @spec make_name(Recipe.t(), atom, atom) :: atom def make_name(recipe = %Recipe{}, bread_id, short_name) do [recipe.name, bread_id, short_name] |> Module.concat() end @spec make_bread_name(Recipe.t(), atom()) :: atom() def make_bread_name(recipe = %Recipe{}, bread_id) do [recipe.name, bread_id] |> Module.concat() end @spec get_name(Supervisor.t(), atom) :: atom() def get_name(supervisor, short_name) do Module.concat([Supervisor.get_root_name(supervisor), short_name]) end @spec concat_name(Bread.t(), atom) :: atom def concat_name(bread = %Bread{}, grain) do Module.concat([bread.name, grain]) end @doc """ Retrieve a grain's substate. This is similar to `:sys.get_state/1`. Note that this function should usually not be used, but it can be useful for debugging or testing. ## Errors The function exits the caller on error. For example, if no bread with the supplied id can be found, this results in a `no_proc` exit. """ @spec get_substate(atom) :: term() def get_substate(bread_id) do bread_id |> :sys.get_state() |> Map.get(:substate) end @doc """ Description of a periodic grain. """ def periodic(grain, period, args \\ %{with_timestamps: false}) do {:periodic, grain, Map.put(args, :period, period)} end @doc """ A ghost edge. A ghost edge is an edge from one grain through another which is not usable by Grains itself, but which may indicate communication between the two nodes using standard tools such as `GenServer.call/2`. """ def ghost(grain, description \\ nil) do {:ghost, grain, description: description} end @doc """ Combine two sets of recipes and grains. """ def combine({ra = %Recipe{}, ga = %Grains{}}, {rb = %Recipe{}, gb = %Grains{}}) do rab = Recipe.merge(ra.name, ra, rb) gab = Grains.merge(ga, gb) {rab, gab} end defdelegate get_original_recipe!(supervisor), to: Grains.Supervisor @doc """ Debug an asynchronous data chain. The function expects a list of grains. The grain currently handling the message checks if its name matches the first element of the list. If so, it pops that element and attempts to send the message to the next grain in the list. If only one element in the list remains, and that element matches the name of the current grain handling the list, this grain replies to the sender with `message`. """ def debug_reply_chain(reply_chain = [first_grain | _], message) do GenServer.call(first_grain, {:debug, :reply_chain, self(), reply_chain, message}) end @doc """ Debug multiple asynchronous data paths. This function floods `message` along all paths between `first_grain` and `last_grain`. It is a generalization of `debug_reply_chain/2`, which does not require specifying the paths explicitly. The function returns `{:ok, number_of_paths}` to indicate how often `message` should be received by the caller of this function. """ def debug_reply_chain(first_grain, last_grain, message) do GenServer.call(first_grain, {:debug, :automatic_reply_chain, last_grain, self(), message}) end @doc """ Debug multiple asynchronous pull paths. This function floods pulls upwards from `first_grain` towards `last_grain`. The function returns `{:ok, number_of_paths}` to indicate how often `message` should be received by the caller of this function. """ def debug_pull_chain(first_grain, last_grain, message) do GenServer.call(first_grain, {:debug, :automatic_pull_chain, last_grain, self(), message}) end @doc """ Define a pattern which messages must match in order to be pushed to the successor grains. Every message gets tested against the pattern before it is pushed to the successors. """ defmacro route(pattern, grains) do build_route(pattern, grains) end @doc """ Convenient version of `routes/2`. Takes a list of tuples of the form `[{pattern, grains}]`. """ defmacro route(routes) when is_list(routes) do Enum.map(routes, fn {l, r} -> build_route(l, r) end) end defp build_route(pattern, grains) do router = quote do fn unquote(pattern) -> true _ -> false end end string = pattern |> Macro.to_string() grains = List.wrap(grains) quote do {:route, unquote(router), unquote(grains), unquote(string)} end end end