diff --git a/test/emqx_shared_sub_SUITE.erl b/test/emqx_shared_sub_SUITE.erl index 30bd96a66..02c4c6598 100644 --- a/test/emqx_shared_sub_SUITE.erl +++ b/test/emqx_shared_sub_SUITE.erl @@ -37,11 +37,18 @@ all() -> emqx_ct:all(?SUITE). init_per_suite(Config) -> net_kernel:start(['master@127.0.0.1', longnames]), emqx_ct_helpers:boot_modules(all), + PortDiscovery = application:get_env(gen_rpc, port_discovery), + application:set_env(gen_rpc, port_discovery, stateless), + application:ensure_all_started(gen_rpc), emqx_ct_helpers:start_apps([]), - Config. + [{port_discovery, PortDiscovery} | Config]. -end_per_suite(_Config) -> - emqx_ct_helpers:stop_apps([]). +end_per_suite(Config) -> + emqx_ct_helpers:stop_apps([gen_rpc]), + case proplists:get_value(port_discovery, Config) of + {ok, OldValue} -> application:set_env(gen_rpc, port_discovery, OldValue); + _ -> ok + end. t_is_ack_required(_) -> ?assertEqual(false, emqx_shared_sub:is_ack_required(#message{headers = #{}})). @@ -284,7 +291,7 @@ test_two_messages(Strategy, Group) -> ok. last_message(ExpectedPayload, Pids) -> - last_message(ExpectedPayload, Pids, 100). + last_message(ExpectedPayload, Pids, 6000). last_message(ExpectedPayload, Pids, Timeout) -> receive @@ -336,7 +343,7 @@ t_per_group_config(_) -> test_two_messages(round_robin, <<"round_robin_group">>). t_local(_) -> - Node = start_slave('local_shared_sub_test', 21884), + Node = start_slave('local_shared_sub_test19', 21884), GroupConfig = #{ <<"local_group_fallback">> => local, <<"local_group">> => local, @@ -392,7 +399,7 @@ t_local_fallback(_) -> Topic = <<"local_foo/bar">>, ClientId1 = <<"ClientId1">>, ClientId2 = <<"ClientId2">>, - Node = start_slave('local_fallback_shared_sub_test', 11885), + Node = start_slave('local_fallback_shared_sub_test19', 11885), {ok, ConnPid1} = emqtt:start_link([{clientid, ClientId1}]), {ok, _} = emqtt:connect(ConnPid1), @@ -401,10 +408,10 @@ t_local_fallback(_) -> emqtt:subscribe(ConnPid1, {<<"$share/local_group_fallback/local_foo/bar">>, 0}), - emqx:publish(Message1), + [{share, <<"local_foo/bar">>, {ok, 1}}] = emqx:publish(Message1), {true, UsedSubPid1} = last_message(<<"hello1">>, [ConnPid1]), - rpc:call(Node, emqx, publish, [Message2]), + [{share, <<"local_foo/bar">>, {ok, 1}}] = rpc:call(Node, emqx, publish, [Message2]), {true, UsedSubPid2} = last_message(<<"hello2">>, [ConnPid1]), emqtt:stop(ConnPid1), @@ -537,7 +544,7 @@ start_slave(Name, Port) -> {erl_flags, ebin_path()}]), pong = net_adm:ping(Node), - setup_node(Node, Port), + ok = setup_node(Node, Port), Node. stop_slave(Node) -> @@ -563,8 +570,7 @@ setup_node(Node, Port) -> name => "internal", opts => [{zone,internal}], proto => tcp}]), - application:set_env(gen_rpc, port_discovery, manual), - application:set_env(gen_rpc, tcp_server_port, Port * 2), + application:set_env(gen_rpc, port_discovery, stateless), ok; (_) -> ok @@ -572,7 +578,13 @@ setup_node(Node, Port) -> [ok = rpc:call(Node, application, load, [App]) || App <- [gen_rpc, emqx]], ok = rpc:call(Node, emqx_ct_helpers, start_apps, [[emqx], EnvHandler]), - rpc:call(Node, ekka, join, [node()]), + %% Sanity check. Assert that `gen_rpc' is set up correctly: + ?assertEqual( Node + , gen_rpc:call(Node, erlang, node, []) + ), + ?assertEqual( node() + , gen_rpc:call(Node, gen_rpc, call, [node(), erlang, node, []]) + ), ok.