partisan_peer_service_connections.erl

%% -------------------------------------------------------------------
%%
%% Copyright (c) 2017 Christopher Meiklejohn.  All Rights Reserved.
%%
%% This file is provided to you under the Apache License,
%% Version 2.0 (the "License"); you may not use this file
%% except in compliance with the License.  You may obtain
%% a copy of the License at
%%
%%   http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing,
%% software distributed under the License is distributed on an
%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
%% KIND, either express or implied.  See the License for the
%% specific language governing permissions and limitations
%% under the License.
%%
%% -------------------------------------------------------------------

%% @doc API for managing peer service connections
-module(partisan_peer_service_connections).

-export([new/0,
         find/2,
         store/3,
         prune/2,
         erase/2,
         foreach/2]).

-include("partisan.hrl").

-type t() :: dict:dict(node_spec(), list(pid())).
-export_type([t/0]).

-type entries() :: [entry()].
-type entry() :: {listen_addr(), channel(), pid()}.

%% @doc Creates a new dictionary of connections.
-spec new() -> t().
new() ->
    dict:new().

%% @doc Finds connection pids in dictionary either by name or node spec.
-spec find(Node :: atom() | node_spec(),
           Connections :: t()) -> {ok, entries()} | {error, not_found}.
find(Name, Connections) when is_atom(Name) ->
    find_first_name(Name, dict:to_list(Connections));
find(Node, Connections) ->
    case dict:find(Node, Connections) of
        error ->
            {error, not_found};
        {ok, Entries} ->
            {ok, Entries}
    end.

%% @doc Store a connection pid
-spec store(Node :: node_spec(),
            Entry :: entry(),
            Connections :: t()) -> t().
store(Node, {_ListenAddr, _Channel, _Pids} = Entry, Connections) ->
    case find(Node, Connections) of
        {error, not_found} ->
            dict:store(Node, [Entry], Connections);
        _ ->
            dict:fold(fun(Node0, Entries, ConnectionsIn) when Node0 =:= Node ->
                              dict:store(Node, Entries ++ [Entry], ConnectionsIn);
                         (_Node, _Entries, ConnectionsIn) ->
                              ConnectionsIn
                      end, Connections, Connections)
    end.

%% @doc Prune all occurrences of a connection pid
%%      returns the node where the pruned pid was found
-spec prune(pid() | node_spec(),
            Connections :: t()) -> {node_spec(), t()}.
prune(#{name := _Name} = Node, Connections) ->
    {Node, dict:erase(Node, Connections)};
prune(Pid, Connections) when is_pid(Pid) ->
    dict:fold(fun(Node, Entries, {AccNode, ConnectionsIn}) ->
                    case lists:keymember(Pid, 3, Entries) of
                        true ->
                            case lists:keydelete(Pid, 3, Entries) of
                                [] ->
                                    {Node, dict:erase(Node, ConnectionsIn)};
                                LeftEntries ->
                                    {Node,
                                     dict:store(Node, LeftEntries, ConnectionsIn)}
                            end;
                        false ->
                            {AccNode, ConnectionsIn}
                    end
              end, {undefined, Connections}, Connections).

erase(NodeName, Connections) ->
    dict:fold(fun(#{name := Name} = Node, {_ListenAddr, _Channel, Pid} = Entry, ConnectionsAcc) ->
                      case Name of
                          NodeName ->
                              try
                                  gen_server:stop(Pid, normal, infinity)
                              catch
                                  _:_ ->
                                      ok
                              end,
                              ConnectionsAcc;
                          _ ->
                              dict:store(Node, Entry, ConnectionsAcc)
                      end
              end, dict:new(), Connections).

%% @doc Apply a function to all connection entries
-spec foreach(Fun :: fun((node_spec(), list(pid())) -> ok),
              Connections :: t()) -> ok.
foreach(Fun, Connections) ->
    _ = dict:map(Fun, Connections),
    ok.

%% @private
-spec find_first_name(Name :: atom(),
                      ConnectionsList :: [{node_spec(), entries()}]) ->
            {ok, entries()} | {error, not_found}.
find_first_name(_Name, []) -> {error, not_found};
find_first_name(Name, [{#{name := Name}, Entries}|_]) -> {ok, Entries};
find_first_name(Name, [_|Rest]) -> find_first_name(Name, Rest).

%%
%% Tests
%%
-ifdef(TEST).

-include_lib("eunit/include/eunit.hrl").

node1() ->
    #{name => node1, listen_addrs => [node1_listen_addr()]}.

node1_listen_addr() ->
    #{ip => {127, 0, 0, 1}, port => 80}.

node2() ->
    #{name => node2, listen_addrs => [node2_listen_addr()]}.

node2_listen_addr() ->
    #{ip => {127, 0, 0, 1}, port => 81}.

node1_bind() ->
    {node1_listen_addr(), undefined, self()}.

node2_bind() ->
    {node2_listen_addr(), undefined, self()}.

no_connections_test() ->
    Connections0 = new(),
    ?assertEqual({error, not_found}, find(node1, Connections0)).

one_connection_test() ->
    Connections0 = new(),
    Connections1 = store(node1(), node1_bind(), Connections0),
    ?assertEqual({ok, [node1_bind()]}, find(node1(), Connections1)),
    Connections2 = store(node2(), node2_bind(), Connections1),
    ?assertEqual({ok, [node2_bind()]}, find(node2, Connections2)).

several_connections_test() ->
    Connections0 = new(),
    Connections1 = store(node1(), node1_bind(), Connections0),
    Connections2 = store(node1(), node1_bind(), Connections1),
    ?assertEqual({ok, [node1_bind(), node1_bind()]}, find(node1(), Connections2)),
    ?assertEqual({ok, [node1_bind(), node1_bind()]}, find(node1, Connections2)).

prune_connections_test() ->
    Connections0 = new(),
    Connections1 = store(node1(), node1_bind(), Connections0),
    Connections2 = store(node1(), node1_bind(), Connections1),
    ?assertEqual({ok, [node1_bind(), node1_bind()]}, find(node1(), Connections2)),
    {#{name := node1, listen_addrs := [#{ip := {127, 0, 0, 1}, port := 80}]},
     Connections3} = prune(self(), Connections2),
    ?assertEqual({ok, [node1_bind()]}, find(node1(), Connections3)),
    {#{name := node1, listen_addrs := [#{ip := {127, 0, 0, 1}, port := 80}]},
     Connections4} = prune(self(), Connections3),
    ?assertEqual({error, not_found}, find(node1(), Connections4)),
    Connections5 = store(node1(), node1_bind(), Connections4),
    Connections6 = store(node1(), node1_bind(), Connections5),
    {#{name := node1, listen_addrs := [#{ip := {127, 0, 0, 1}, port := 80}]},
     Connections7} = prune(self(), Connections6),
    ?assertEqual({ok, [node1_bind()]}, find(node1, Connections7)),
    {#{name := node1, listen_addrs := [#{ip := {127, 0, 0, 1}, port := 80}]},
     Connections8} = prune(self(), Connections7),
    ?assertEqual({error, not_found}, find(node1, Connections8)).

add_remove_add_connection_test() ->
    Connections0 = new(),
    Connections1 = store(node1(), node1_bind(), Connections0),
    ?assertEqual({ok, [node1_bind()]}, find(node1, Connections1)),
    {#{name := node1, listen_addrs := [#{ip := {127, 0, 0, 1}, port := 80}]},
     Connections2} = prune(self(), Connections1),
    ?assertEqual({error, not_found}, find(node1, Connections2)),
    Connections3 = store(node1(), node1_bind(), Connections2),
    ?assertEqual({ok, [node1_bind()]}, find(node1, Connections3)),
    {#{name := node1, listen_addrs := [#{ip := {127, 0, 0, 1}, port := 80}]},
     Connections4} = prune(node1(), Connections3),
    ?assertEqual({error, not_found}, find(node1, Connections4)).

-endif.