# # Distribunator.Manager defmodule Distribunator.Manager do @moduledoc """ The Distribunator.Manager: 1. Makes sure all distributed nodes are `Node.connect()`ed """ use GenServer require Distribunator.Distributed ############################## # API ############################## @doc """ Start the genserver with a list of nodes to connect to and monitor ## Parameters - `absent_nodes` The list of nodes (each passable to `Node.connect`) that this GenServer will connect to and monitor ## Returns - See `GenServer::on_start` """ def start_link(absent_nodes \\ []) do GenServer.start_link(__MODULE__, absent_nodes, [name: __MODULE__]) end @doc """ Connect to one or more nodes that are not currently connected or monitored ## Parameters - `nodes` A single node or list of node atoms to connect and monitor ## Returns - `:ok` All is well, the nodes are being connected and monitored - `{:error, reason}` Failed for reason """ def connect(nodes), do: Distribunator.Distributed.call({:connect, nodes}) @doc """ Disconnect from one or more nodes that are currently connected and monitored ## Parameters - `nodes` A single node or list of node atoms to disconnect ## Returns - `:ok` All is well, the nodes are disconnected - `{:error, reason}` Failed for reason """ def disconnect(nodes), do: Distribunator.Distributed.call({:disconnect, nodes}) defmodule State do @moduledoc false defstruct [ absent_nodes: [], # list of nodes we're not connected to but should be present_nodes: [] # list nodes we are connected to and monitoring ] end ############################## # GenServer Callbacks ############################## @impl GenServer def init(absent_nodes) do Distribunator.Distributed.register() send(self(), :reconcile) {:ok, %State{absent_nodes: absent_nodes}} end @impl GenServer def handle_call({:connect, nodes}, _from, state) do {updated_state, result} = do_connect(state, List.wrap(nodes)) {:reply, result, updated_state} end @impl GenServer def handle_call({:disconnect, nodes}, _from, state) do {updated_state, result} = do_disconnect(state, List.wrap(nodes)) {:reply, result, updated_state} end @impl GenServer def handle_info(:reconcile, state) do {:noreply, do_reconcile(state)} end @impl GenServer def handle_info({:nodedown, n}, state) do new_absent_nodes = [n| state.absent_nodes] |> Enum.uniq() new_present_nodes = Enum.reject(state.present_nodes, n) Process.send_after(self(), :reconcile, 5_000) new_state = state |> Map.put(:absent_nodes, new_absent_nodes) |> Map.put(:present_nodes, new_present_nodes) {:noreply, new_state} end ############################## # Internal Calls ############################## def do_reconcile(%State{absent_nodes: [] } = state), do: state def do_reconcile(%State{absent_nodes: absent_nodes, present_nodes: present_nodes} = state) do {new_absent_nodes, new_present_nodes} = Enum.reduce(absent_nodes, {[], present_nodes}, fn(absent_node, {acc_absent, acc_present}) -> if Node.connect(absent_node) && Node.monitor(absent_node, true) do # was previously absent, is now present {acc_absent, [absent_node| acc_present]} else # was previously absent, still is absent {[absent_node| acc_absent], acc_present} end end) if Enum.any?(new_absent_nodes) do # try again in a little while Process.send_after(self(), :reconcile, 5_000) end state |> Map.put(:absent_nodes, new_absent_nodes) |> Map.put(:present_nodes, new_present_nodes) end def do_connect(%State{absent_nodes: absent_nodes, present_nodes: present_nodes} = state, new_nodes) do if Enum.any?(new_nodes, fn(node) -> node in absent_nodes or node in present_nodes end) do {state, {:error, "duplicate node"}} else send(self(), :reconcile) {Map.put(state, :absent_nodes, absent_nodes ++ new_nodes), :ok} end end def do_disconnect(%State{absent_nodes: absent_nodes, present_nodes: present_nodes} = state, disconnect_nodes) do if not Enum.all?(disconnect_nodes, fn(node) -> node in absent_nodes or node in present_nodes end) do {state, {:error, "no node"}} else for node <- disconnect_nodes do Node.monitor(node, false) Node.disconnect(node) end updated_state = state |> Map.put(:absent_nodes, absent_nodes -- disconnect_nodes) |> Map.put(:present_nodes, present_nodes -- disconnect_nodes) {updated_state, :ok} end end end