defmodule Adap.Unit do @moduledoc "Behaviour describing an ADAP distributed processing unit" use Behaviour defcallback start_link(args :: term) :: {:ok,pid} defcallback cast(pid,fun) :: :ok defcallback node(args :: term) :: node end defmodule Adap.Unit.Router do @moduledoc """ Route element to a node/process started on demand: `Adap.Unit.Router.cast({mod,arg}=unit_spec,elem)` will: - route the query to `mod.node(arg)` - see if a process for the spec `{mod,arg}` is running locally - if not start a process tree with `mod.start_link(arg)` - route the query to existing or newly created process with `mod.cast(pid,elem)` Processes are monitored in order to restart them on demand when they die. A process specification is defined as a tuple `{module,args}`: module must implement behaviour `Adap.Unit` with previously described callbacks. A Unit can represent : a GenServer, a pool of GenServers, a pool of node of GenServer, etc. The reference unit is a simple GenServer: - which dies itself after a given "time to live" - where the routed element is an anonymous function with one parameter - casting the function on server and apply it with the server state as parameter You can `use Adap.Unit.Simple` to take the default implementation for this kind of processing unit. """ use GenServer def start_link, do: GenServer.start_link(__MODULE__,[], name: __MODULE__) def cast({m,a},fun), do: GenServer.cast({__MODULE__,m.node(a)},{:route,{m,a},fun}) def init(_), do: {:ok,%{pids: HashDict.new,specs: HashDict.new}} def handle_cast({:route,{m,a}=spec,fun},%{pids: pids,specs: specs}=state) do if (pid=Dict.get(pids,spec)) do m.cast(pid,fun); {:noreply,state} else {:ok,pid} = m.start_link(a) m.cast(pid,fun) {:noreply,%{state| pids: Dict.put(pids,spec,pid), specs: Dict.put(specs,pid,spec)}} end end def handle_info({:EXIT, pid, _},%{pids: pids,specs: specs}=state), do: # no need to supervise backends, since they will be restarted by next query {:noreply,%{state|pids: Dict.delete(pids,Dict.fetch!(specs,pid)), specs: Dict.delete(specs,pid)}} def terminate(_,%{pids: pids}), do: Enum.each(pids,fn {_,pid}->Process.exit(pid,:shutdown) end) end defmodule Adap.Unit.Simple do defmacro __using__(opts) do quote do @behaviour Adap.Unit use GenServer def start_link(arg), do: GenServer.start_link(__MODULE__,arg) def cast(pid,fun), do: GenServer.cast(pid,{:apply,fun}) def node(_), do: node def handle_cast({:apply,fun},state), do: (fun.(state); {:noreply,state,unquote(opts[:ttl])}) def handle_info(:timeout,state), do: {:stop,:normal,state} end end end