syn/test/syn_groups_SUITE.erl

%% ==========================================================================================================
%% Syn - A global Process Registry and Process Group manager.
%%
%% The MIT License (MIT)
%%
%% Copyright (c) 2016 Roberto Ostinelli <roberto@ostinelli.net> and Neato Robotics, Inc.
%%
%% Permission is hereby granted, free of charge, to any person obtaining a copy
%% of this software and associated documentation files (the "Software"), to deal
%% in the Software without restriction, including without limitation the rights
%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
%% copies of the Software, and to permit persons to whom the Software is
%% furnished to do so, subject to the following conditions:
%%
%% The above copyright notice and this permission notice shall be included in
%% all copies or substantial portions of the Software.
%%
%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
%% THE SOFTWARE.
%% ==========================================================================================================
-module(syn_groups_SUITE).

%% callbacks
-export([all/0]).
-export([init_per_suite/1, end_per_suite/1]).
-export([groups/0, init_per_group/2, end_per_group/2]).
-export([init_per_testcase/2, end_per_testcase/2]).

%% tests
-export([
    single_node_join/1,
    single_node_leave/1,
    single_node_kill/1,
    single_node_leave_and_kill_multi_groups/1,
    single_node_publish/1,
    single_node_multi_call/1,
    single_node_multi_call_when_recipient_crashes/1,
    single_node_meta/1,
    single_node_callback_on_process_exit/1,
    single_node_all_keys/1
]).
-export([
    two_nodes_kill/1,
    two_nodes_publish/1,
    two_nodes_multi_call/1,
    two_nodes_local_members/1,
    two_nodes_local_publish/1
]).

%% internals
-export([recipient_loop/1]).
-export([called_loop/1, called_loop_that_crashes/1]).
-export([process_groups_process_exit_callback_dummy/4]).

%% include
-include_lib("common_test/include/ct.hrl").


%% ===================================================================
%% Callbacks
%% ===================================================================

%% -------------------------------------------------------------------
%% Function: all() -> GroupsAndTestCases | {skip,Reason}
%% GroupsAndTestCases = [{group,GroupName} | TestCase]
%% GroupName = atom()
%% TestCase = atom()
%% Reason = term()
%% -------------------------------------------------------------------
all() ->
    [
        {group, single_node_process_groups},
        {group, two_nodes_process_groups}
    ].

%% -------------------------------------------------------------------
%% Function: groups() -> [Group]
%% Group = {GroupName,Properties,GroupsAndTestCases}
%% GroupName = atom()
%% Properties = [parallel | sequence | Shuffle | {RepeatType,N}]
%% GroupsAndTestCases = [Group | {group,GroupName} | TestCase]
%% TestCase = atom()
%% Shuffle = shuffle | {shuffle,{integer(),integer(),integer()}}
%% RepeatType = repeat | repeat_until_all_ok | repeat_until_all_fail |
%%			   repeat_until_any_ok | repeat_until_any_fail
%% N = integer() | forever
%% -------------------------------------------------------------------
groups() ->
    [
        {single_node_process_groups, [shuffle], [
            single_node_join,
            single_node_leave,
            single_node_kill,
            single_node_leave_and_kill_multi_groups,
            single_node_publish,
            single_node_multi_call,
            single_node_multi_call_when_recipient_crashes,
            single_node_meta,
            single_node_callback_on_process_exit,
            single_node_all_keys
        ]},
        {two_nodes_process_groups, [shuffle], [
            two_nodes_kill,
            two_nodes_publish,
            two_nodes_multi_call,
            two_nodes_local_members,
            two_nodes_local_publish
        ]}
    ].
%% -------------------------------------------------------------------
%% Function: init_per_suite(Config0) ->
%%				Config1 | {skip,Reason} |
%%              {skip_and_save,Reason,Config1}
%% Config0 = Config1 = [tuple()]
%% Reason = term()
%% -------------------------------------------------------------------
init_per_suite(Config) ->
    %% config
    [
        {slave_node_short_name, syn_slave}
        | Config
    ].

%% -------------------------------------------------------------------
%% Function: end_per_suite(Config0) -> void() | {save_config,Config1}
%% Config0 = Config1 = [tuple()]
%% -------------------------------------------------------------------
end_per_suite(_Config) -> ok.

%% -------------------------------------------------------------------
%% Function: init_per_group(GroupName, Config0) ->
%%				Config1 | {skip,Reason} |
%%              {skip_and_save,Reason,Config1}
%% GroupName = atom()
%% Config0 = Config1 = [tuple()]
%% Reason = term()
%% -------------------------------------------------------------------
init_per_group(two_nodes_process_groups, Config) ->
    %% start slave
    SlaveNodeShortName = proplists:get_value(slave_node_short_name, Config),
    {ok, SlaveNode} = syn_test_suite_helper:start_slave(SlaveNodeShortName),
    %% config
    [
        {slave_node, SlaveNode}
        | Config
    ];
init_per_group(_GroupName, Config) -> Config.

%% -------------------------------------------------------------------
%% Function: end_per_group(GroupName, Config0) ->
%%				void() | {save_config,Config1}
%% GroupName = atom()
%% Config0 = Config1 = [tuple()]
%% -------------------------------------------------------------------
end_per_group(two_nodes_process_groups, Config) ->
    %% get slave node name
    SlaveNodeShortName = proplists:get_value(slave_node_short_name, Config),
    %% stop slave
    syn_test_suite_helper:stop_slave(SlaveNodeShortName);
end_per_group(_GroupName, _Config) ->
    ok.

% ----------------------------------------------------------------------------------------------------------
% Function: init_per_testcase(TestCase, Config0) ->
%				Config1 | {skip,Reason} | {skip_and_save,Reason,Config1}
% TestCase = atom()
% Config0 = Config1 = [tuple()]
% Reason = term()
% ----------------------------------------------------------------------------------------------------------
init_per_testcase(_TestCase, Config) ->
    Config.

% ----------------------------------------------------------------------------------------------------------
% Function: end_per_testcase(TestCase, Config0) ->
%				void() | {save_config,Config1} | {fail,Reason}
% TestCase = atom()
% Config0 = Config1 = [tuple()]
% Reason = term()
% ----------------------------------------------------------------------------------------------------------
end_per_testcase(_TestCase, Config) ->
    %% get slave
    SlaveNode = proplists:get_value(slave_node, Config),
    syn_test_suite_helper:clean_after_test(SlaveNode).

%% ===================================================================
%% Tests
%% ===================================================================
single_node_join(_Config) ->
    %% set schema location
    application:set_env(mnesia, schema_location, ram),
    %% start
    ok = syn:start(),
    ok = syn:init(),
    %% start process
    Pid = syn_test_suite_helper:start_process(),
    %% retrieve
    [] = syn:get_members(<<"my group">>),
    false = syn:member(Pid, <<"my group">>),
    %% join
    ok = syn:join(<<"my group">>, Pid),
    %% allow to rejoin
    ok = syn:join(<<"my group">>, Pid),
    %% retrieve
    [Pid] = syn:get_members(<<"my group">>),
    true = syn:member(Pid, <<"my group">>),
    %% kill process
    syn_test_suite_helper:kill_process(Pid).

single_node_leave(_Config) ->
    %% set schema location
    application:set_env(mnesia, schema_location, ram),
    %% start
    ok = syn:start(),
    ok = syn:init(),
    %% start process
    Pid = syn_test_suite_helper:start_process(),
    %% retrieve
    [] = syn:get_members(<<"my group">>),
    false = syn:member(Pid, <<"my group">>),
    %% leave before join
    {error, pid_not_in_group} = syn:leave(<<"my group">>, Pid),
    %% join
    ok = syn:join(<<"my group">>, Pid),
    %% retrieve
    [Pid] = syn:get_members(<<"my group">>),
    true = syn:member(Pid, <<"my group">>),
    %% leave
    ok = syn:leave(<<"my group">>, Pid),
    %% retrieve
    [] = syn:get_members(<<"my group">>),
    false = syn:member(Pid, <<"my group">>),
    %% kill process
    syn_test_suite_helper:kill_process(Pid).

single_node_kill(_Config) ->
    %% set schema location
    application:set_env(mnesia, schema_location, ram),
    %% start
    ok = syn:start(),
    ok = syn:init(),
    %% start process
    Pid = syn_test_suite_helper:start_process(),
    %% retrieve
    [] = syn:get_members(<<"my group 1">>),
    [] = syn:get_members(<<"my group 2">>),
    false = syn:member(Pid, <<"my group 1">>),
    false = syn:member(Pid, <<"my group 2">>),
    %% join
    ok = syn:join(<<"my group 1">>, Pid),
    ok = syn:join(<<"my group 2">>, Pid),
    %% retrieve
    [Pid] = syn:get_members(<<"my group 1">>),
    [Pid] = syn:get_members(<<"my group 2">>),
    true = syn:member(Pid, <<"my group 1">>),
    true = syn:member(Pid, <<"my group 2">>),
    %% kill process
    syn_test_suite_helper:kill_process(Pid),
    timer:sleep(100),
    %% retrieve
    [] = syn:get_members(<<"my group 1">>),
    [] = syn:get_members(<<"my group 2">>),
    false = syn:member(Pid, <<"my group 1">>),
    false = syn:member(Pid, <<"my group 2">>).

single_node_leave_and_kill_multi_groups(_Config) ->
    %% set schema location
    application:set_env(mnesia, schema_location, ram),
    %% start
    ok = syn:start(),
    ok = syn:init(),
    %% start process
    Pid = syn_test_suite_helper:start_process(),
    %% retrieve
    [] = syn:get_members(<<"my group 1">>),
    [] = syn:get_members(<<"my group 2">>),
    false = syn:member(Pid, <<"my group 1">>),
    false = syn:member(Pid, <<"my group 2">>),
    %% join
    ok = syn:join(<<"my group 1">>, Pid),
    ok = syn:join(<<"my group 2">>, Pid),
    %% retrieve
    [Pid] = syn:get_members(<<"my group 1">>),
    [Pid] = syn:get_members(<<"my group 2">>),
    true = syn:member(Pid, <<"my group 1">>),
    true = syn:member(Pid, <<"my group 2">>),
    %% leave group 1
    ok = syn:leave(<<"my group 1">>, Pid),
    %% retrieve
    [] = syn:get_members(<<"my group 1">>),
    [Pid] = syn:get_members(<<"my group 2">>),
    false = syn:member(Pid, <<"my group 1">>),
    true = syn:member(Pid, <<"my group 2">>),
    %% kill process
    syn_test_suite_helper:kill_process(Pid),
    timer:sleep(100),
    %% retrieve
    [] = syn:get_members(<<"my group 1">>),
    [] = syn:get_members(<<"my group 2">>),
    false = syn:member(Pid, <<"my group 1">>),
    false = syn:member(Pid, <<"my group 2">>).

single_node_publish(_Config) ->
    %% set schema location
    application:set_env(mnesia, schema_location, ram),
    %% start
    ok = syn:start(),
    ok = syn:init(),
    %% start processes
    ResultPid = self(),
    F = fun() -> recipient_loop(ResultPid) end,
    Pid1 = syn_test_suite_helper:start_process(F),
    Pid2 = syn_test_suite_helper:start_process(F),
    %% join
    ok = syn:join(<<"my group">>, Pid1),
    ok = syn:join(<<"my group">>, Pid2),
    %% publish
    {ok, 2} = syn:publish(<<"my group">>, {test, message}),
    %% check publish was received
    receive
        {received, Pid1, {test, message}} -> ok
    after 2000 ->
        ok = published_message_was_not_received_by_pid1
    end,
    receive
        {received, Pid2, {test, message}} -> ok
    after 2000 ->
        ok = published_message_was_not_received_by_pid2
    end,
    %% kill processes
    syn_test_suite_helper:kill_process(Pid1),
    syn_test_suite_helper:kill_process(Pid2).

single_node_multi_call(_Config) ->
    %% set schema location
    application:set_env(mnesia, schema_location, ram),
    %% start
    ok = syn:start(),
    ok = syn:init(),
    %% start processes
    Pid1 = syn_test_suite_helper:start_process(fun() -> called_loop(pid1) end),
    Pid2 = syn_test_suite_helper:start_process(fun() -> called_loop(pid2) end),
    PidUnresponsive = syn_test_suite_helper:start_process(),
    %% register
    ok = syn:join(<<"my group">>, Pid1),
    ok = syn:join(<<"my group">>, Pid2),
    ok = syn:join(<<"my group">>, PidUnresponsive),
    %% call
    {Replies, BadPids} = syn:multi_call(<<"my group">>, get_pid_name),
    %% check responses
    2 = length(Replies),
    pid1 = proplists:get_value(Pid1, Replies),
    pid2 = proplists:get_value(Pid2, Replies),
    [PidUnresponsive] = BadPids,
    %% kill processes
    syn_test_suite_helper:kill_process(Pid1),
    syn_test_suite_helper:kill_process(Pid2),
    syn_test_suite_helper:kill_process(PidUnresponsive).

single_node_multi_call_when_recipient_crashes(_Config) ->
    %% set schema location
    application:set_env(mnesia, schema_location, ram),
    %% start
    ok = syn:start(),
    ok = syn:init(),
    %% start processes
    Pid1 = syn_test_suite_helper:start_process(fun() -> called_loop(pid1) end),
    Pid2 = syn_test_suite_helper:start_process(fun() -> called_loop(pid2) end),
    PidCrashes = syn_test_suite_helper:start_process(fun() -> called_loop_that_crashes(pid_crashes) end),
    %% register
    ok = syn:join(<<"my group">>, Pid1),
    ok = syn:join(<<"my group">>, Pid2),
    ok = syn:join(<<"my group">>, PidCrashes),
    %% call
    {Time, {Replies, BadPids}} = timer:tc(syn, multi_call, [<<"my group">>, get_pid_name]),
    %% check that pid2 was monitored, no need to wait for timeout
    true = Time / 1000 < 1000,
    %% check responses
    2 = length(Replies),
    pid1 = proplists:get_value(Pid1, Replies),
    pid2 = proplists:get_value(Pid2, Replies),
    [PidCrashes] = BadPids,
    %% kill processes
    syn_test_suite_helper:kill_process(Pid1),
    syn_test_suite_helper:kill_process(Pid2).

single_node_meta(_Config) ->
    %% set schema location
    application:set_env(mnesia, schema_location, ram),
    %% start
    ok = syn:start(),
    ok = syn:init(),
    %% start process
    Pid = syn_test_suite_helper:start_process(),
    %% retrieve
    [] = syn:get_members(<<"my group">>, with_meta),
    false = syn:member(Pid, <<"my group">>),
    %% join
    ok = syn:join(<<"my group">>, Pid, {some, meta}),
    %% retrieve
    [{Pid, {some, meta}}] = syn:get_members(<<"my group">>, with_meta),
    %% allow to rejoin to update meta
    ok = syn:join(<<"my group">>, Pid, {updated, meta}),
    %% retrieve
    [{Pid, {updated, meta}}] = syn:get_members(<<"my group">>, with_meta),
    %% leave
    ok = syn:leave(<<"my group">>, Pid),
    %% retrieve
    [] = syn:get_members(<<"my group">>),
    false = syn:member(Pid, <<"my group">>),
    %% kill process
    syn_test_suite_helper:kill_process(Pid).

single_node_callback_on_process_exit(_Config) ->
    CurrentNode = node(),
    %% set schema location
    application:set_env(mnesia, schema_location, ram),
    %% load configuration variables from syn-test.config => this defines the callback
    syn_test_suite_helper:set_environment_variables(),
    %% start
    ok = syn:start(),
    ok = syn:init(),
    %% register global process
    ResultPid = self(),
    global:register_name(syn_process_groups_SUITE_result, ResultPid),
    %% start process
    Pid = syn_test_suite_helper:start_process(),
    %% register
    ok = syn:join(<<"my group">>, Pid, {some, meta, 1}),
    ok = syn:join(<<"my other group">>, Pid, {some, meta, 2}),
    %% kill process
    syn_test_suite_helper:kill_process(Pid),
    %% check callback were triggered
    receive
        {exited, CurrentNode, <<"my group">>, Pid, {some, meta, 1}, killed} -> ok
    after 2000 ->
        ok = process_groups_exit_callback_was_not_called_from_local_node
    end,
    receive
        {exited, CurrentNode, <<"my other group">>, Pid, {some, meta, 2}, killed} -> ok
    after 2000 ->
        ok = process_groups_exit_callback_was_not_called_from_local_node
    end,
    %% unregister
    global:unregister_name(syn_process_groups_SUITE_result).

%% covering bug <https://github.com/ostinelli/syn/issues/32>
single_node_all_keys(_Config) ->
    %% set schema location
    application:set_env(mnesia, schema_location, ram),
    %% start
    ok = syn:start(),
    ok = syn:init(),
    %% start process
    Pid = syn_test_suite_helper:start_process(),
    %% join with complex names
    GroupNames = [
        "mygroup",
        <<"mygroup">>,
        mygroup,
        {mygroup},
        {mygroup, 12345},
        {mygroup, {other, <<"mygroup">>}},
        [mygroup, {other, <<"mygroup">>}],
        {mygroup, {other, <<"mygroup">>}, [mygroup, {other, <<"mygroup">>}]}
    ],
    F = fun(GroupName) ->
        ok = syn:join(GroupName, Pid),
        %% retrieve
        [Pid] = syn:get_members(GroupName),
        true = syn:member(Pid, GroupName)
    end,
    lists:foreach(F, GroupNames),
    %% kill process
    syn_test_suite_helper:kill_process(Pid).

two_nodes_kill(Config) ->
    %% get slave
    SlaveNode = proplists:get_value(slave_node, Config),
    %% set schema location
    application:set_env(mnesia, schema_location, ram),
    rpc:call(SlaveNode, mnesia, schema_location, [ram]),
    %% start
    ok = syn:start(),
    ok = syn:init(),
    ok = rpc:call(SlaveNode, syn, start, []),
    ok = rpc:call(SlaveNode, syn, init, []),
    timer:sleep(100),
    %% start processes
    PidLocal = syn_test_suite_helper:start_process(),
    PidSlave = syn_test_suite_helper:start_process(SlaveNode),
    %% retrieve
    [] = syn:get_members(<<"my group">>),
    false = syn:member(PidLocal, <<"my group">>),
    false = syn:member(PidSlave, <<"my group">>),
    [] = rpc:call(SlaveNode, syn, get_members, [<<"my group">>]),
    false = rpc:call(SlaveNode, syn, member, [PidLocal, <<"my group">>]),
    false = rpc:call(SlaveNode, syn, member, [PidSlave, <<"my group">>]),
    %% register
    ok = syn:join(<<"my group">>, PidSlave),
    ok = rpc:call(SlaveNode, syn, join, [<<"my group">>, PidLocal]),
    %% retrieve, pid should have the same order in all nodes
    [PidSlave, PidLocal] = syn:get_members(<<"my group">>),
    [PidSlave, PidLocal] = rpc:call(SlaveNode, syn, get_members, [<<"my group">>]),
    %% kill processes
    syn_test_suite_helper:kill_process(PidLocal),
    syn_test_suite_helper:kill_process(PidSlave),
    timer:sleep(100),
    %% retrieve
    [] = syn:get_members(<<"my group">>),
    false = syn:member(PidLocal, <<"my group">>),
    false = syn:member(PidSlave, <<"my group">>),
    [] = rpc:call(SlaveNode, syn, get_members, [<<"my group">>]),
    false = rpc:call(SlaveNode, syn, member, [PidLocal, <<"my group">>]),
    false = rpc:call(SlaveNode, syn, member, [PidSlave, <<"my group">>]).

two_nodes_publish(Config) ->
    %% get slave
    SlaveNode = proplists:get_value(slave_node, Config),
    %% set schema location
    application:set_env(mnesia, schema_location, ram),
    rpc:call(SlaveNode, mnesia, schema_location, [ram]),
    %% start
    ok = syn:start(),
    ok = syn:init(),
    ok = rpc:call(SlaveNode, syn, start, []),
    ok = rpc:call(SlaveNode, syn, init, []),
    timer:sleep(100),
    %% start process
    ResultPid = self(),
    F = fun() -> recipient_loop(ResultPid) end,
    PidLocal = syn_test_suite_helper:start_process(F),
    PidSlave = syn_test_suite_helper:start_process(SlaveNode, F),
    %% register
    ok = syn:join(<<"my group">>, PidSlave),
    ok = rpc:call(SlaveNode, syn, join, [<<"my group">>, PidLocal]),
    %% publish
    syn:publish(<<"my group">>, {test, message}),
    %% check publish was received
    receive
        {received, PidLocal, {test, message}} -> ok
    after 2000 ->
        ok = published_message_was_not_received_by_pidlocal
    end,
    receive
        {received, PidSlave, {test, message}} -> ok
    after 2000 ->
        ok = published_message_was_not_received_by_pidslave
    end,
    %% kill processes
    syn_test_suite_helper:kill_process(PidLocal),
    syn_test_suite_helper:kill_process(PidSlave).

two_nodes_multi_call(Config) ->
    %% get slave
    SlaveNode = proplists:get_value(slave_node, Config),
    %% set schema location
    application:set_env(mnesia, schema_location, ram),
    rpc:call(SlaveNode, mnesia, schema_location, [ram]),
    %% start
    ok = syn:start(),
    ok = syn:init(),
    ok = rpc:call(SlaveNode, syn, start, []),
    ok = rpc:call(SlaveNode, syn, init, []),
    timer:sleep(100),
    %% start processes
    PidLocal = syn_test_suite_helper:start_process(fun() -> called_loop(pid1) end),
    PidSlave = syn_test_suite_helper:start_process(SlaveNode, fun() -> called_loop(pid2) end),
    PidUnresponsive = syn_test_suite_helper:start_process(),
    %% join
    ok = syn:join(<<"my group">>, PidLocal),
    ok = syn:join(<<"my group">>, PidSlave),
    ok = syn:join(<<"my group">>, PidUnresponsive),
    timer:sleep(100),
    %% call
    {Replies, BadPids} = syn:multi_call(<<"my group">>, get_pid_name, 3000),
    %% check responses
    2 = length(Replies),
    pid1 = proplists:get_value(PidLocal, Replies),
    pid2 = proplists:get_value(PidSlave, Replies),
    [PidUnresponsive] = BadPids,
    %% kill processes
    syn_test_suite_helper:kill_process(PidLocal),
    syn_test_suite_helper:kill_process(PidSlave),
    syn_test_suite_helper:kill_process(PidUnresponsive).

two_nodes_local_members(Config) ->
    %% get slave
    SlaveNode = proplists:get_value(slave_node, Config),
    %% set schema location
    application:set_env(mnesia, schema_location, ram),
    rpc:call(SlaveNode, mnesia, schema_location, [ram]),
    %% start
    ok = syn:start(),
    ok = syn:init(),
    ok = rpc:call(SlaveNode, syn, start, []),
    ok = rpc:call(SlaveNode, syn, init, []),
    timer:sleep(100),
    %% start processes
    PidLocal1 = syn_test_suite_helper:start_process(),
    PidLocal2 = syn_test_suite_helper:start_process(),
    PidSlave = syn_test_suite_helper:start_process(SlaveNode),
    %% join
    ok = syn:join(<<"my group">>, PidLocal1, {meta, pid_local_1}),
    ok = syn:join(<<"my group">>, PidLocal2, {meta, pid_local_2}),
    ok = syn:join(<<"my group">>, PidSlave, {meta, pid_slave}),
    timer:sleep(100),
    %% retrieve, pid should have the same order in all nodes
    [PidLocal1, PidLocal2] = syn:get_local_members(<<"my group">>),
    [
        {PidLocal1, {meta, pid_local_1}},
        {PidLocal2, {meta, pid_local_2}}
    ] = syn:get_local_members(<<"my group">>, with_meta),
    %% local pids leave
    ok = syn:leave(<<"my group">>, PidLocal1),
    ok = syn:leave(<<"my group">>, PidLocal2),
    %% retrieve, no more local pids
    [] = syn:get_local_members(<<"my group">>),
    %% kill processes
    syn_test_suite_helper:kill_process(PidLocal1),
    syn_test_suite_helper:kill_process(PidLocal2),
    syn_test_suite_helper:kill_process(PidSlave).

two_nodes_local_publish(Config) ->
    %% get slave
    SlaveNode = proplists:get_value(slave_node, Config),
    %% set schema location
    application:set_env(mnesia, schema_location, ram),
    rpc:call(SlaveNode, mnesia, schema_location, [ram]),
    %% start
    ok = syn:start(),
    ok = syn:init(),
    ok = rpc:call(SlaveNode, syn, start, []),
    ok = rpc:call(SlaveNode, syn, init, []),
    timer:sleep(100),
    %% start processes
    ResultPid = self(),
    F = fun() -> recipient_loop(ResultPid) end,
    PidLocal1 = syn_test_suite_helper:start_process(F),
    PidLocal2 = syn_test_suite_helper:start_process(F),
    PidSlave = syn_test_suite_helper:start_process(SlaveNode, F),
    %% join
    ok = syn:join(<<"my group">>, PidLocal1, {meta, pid_local_1}),
    ok = syn:join(<<"my group">>, PidLocal2, {meta, pid_local_2}),
    ok = syn:join(<<"my group">>, PidSlave, {meta, pid_slave}),
    %% publish
    {ok, 2} = syn:publish_to_local(<<"my group">>, {test, message}),
    %% check publish was received by local pids
    receive
        {received, PidLocal1, {test, message}} -> ok
    after 2000 ->
        ok = published_message_was_not_received_by_pid_local_1
    end,
    receive
        {received, PidLocal2, {test, message}} -> ok
    after 2000 ->
        ok = published_message_was_not_received_by_pid_local_2
    end,
    receive
        {received, PidSlave, {test, message}} ->
            ko = published_message_was_received_by_pid_slave
    after 1000 ->
        ok
    end,
    %% kill processes
    syn_test_suite_helper:kill_process(PidLocal1),
    syn_test_suite_helper:kill_process(PidLocal2),
    syn_test_suite_helper:kill_process(PidSlave).

%% ===================================================================
%% Internal
%% ===================================================================
recipient_loop(Pid) ->
    receive
        Message -> Pid ! {received, self(), Message}
    end.

called_loop(PidName) ->
    receive
        {syn_multi_call, CallerPid, get_pid_name} -> syn:multi_call_reply(CallerPid, PidName)
    end.

called_loop_that_crashes(_PidName) ->
    receive
        {syn_multi_call, _CallerPid, get_pid_name} -> exit(recipient_crashed_on_purpose)
    end.

process_groups_process_exit_callback_dummy(Name, Pid, Meta, Reason) ->
    global:send(syn_process_groups_SUITE_result, {exited, node(), Name, Pid, Meta, Reason}).