Merge pull request #3959 from esl/c2s-code-cleanup

C2s code cleanup

big_tests/tests/last_SUITE.erl

@@ -57,7 +57,7 @@ init_per_group(valid_queries, Config0) ->
     Config1 = escalus_fresh:create_users(Config0, [{alice, 1}, {bob, 1}]),
     Config2 = escalus:make_everyone_friends(Config1),
     %% This check ensures that there are no registered sessions.
-    %% But in ejabberd_c2s we first unset session,
+    %% But in mongoose_c2s we first unset session,
     %% then broadcast presence unavailable.
     %% This check uses ejabberd_sm to get information about sessions.
     escalus_ejabberd:wait_for_session_count(Config2, 0),

big_tests/tests/mod_global_distrib_SUITE.erl

@@ -409,12 +409,6 @@ test_pm_between_users_before_available_presence(Config) ->
     escalus_client:stop(Config1, Eve).
 
 test_two_way_pm(Alice, Eve) ->
-    %% Ensure that users are properly registered
-    %% Otherwise you can get "Unable to route global message... user not found in the routing table"
-    %% error, because "escalus_client:start" can return before SM registration is completed.
-    wait_for_registration(Alice, ct:get_config({hosts, mim, node})),
-    wait_for_registration(Eve, ct:get_config({hosts, reg, node})),
-
     escalus_client:send(Alice, escalus_stanza:chat_to(Eve, <<"Hi to Eve from Europe1!">>)),
     escalus_client:send(Eve, escalus_stanza:chat_to(Alice, <<"Hi to Alice from Asia!">>)),
 
@@ -443,10 +437,14 @@ test_muc_conversation_on_one_host(Config0) ->
               RoomAddr = muc_helper:room_address(RoomJid),
 
               escalus:send(Alice, muc_helper:stanza_muc_enter_room(RoomJid, AliceUsername)),
-              escalus:wait_for_stanza(Alice),
+              wait_for_muc_presence(Alice, RoomJid, AliceUsername),
+              wait_for_subject(Alice),
 
               escalus:send(Eve, muc_helper:stanza_muc_enter_room(RoomJid, EveUsername)),
-              [_, _, _] = escalus:wait_for_stanzas(Eve, 3),
+              wait_for_muc_presence(Eve, RoomJid, AliceUsername),
+              wait_for_muc_presence(Eve, RoomJid, EveUsername),
+              wait_for_muc_presence(Alice, RoomJid, EveUsername),
+              wait_for_subject(Eve),
 
               Msg= <<"Hi, Eve!">>,
               escalus:send(Alice, escalus_stanza:groupchat_to(RoomAddr, Msg)),
@@ -473,7 +471,7 @@ test_muc_conversation_history(Config0) ->
               RoomAddr = muc_helper:room_address(RoomJid),
 
               escalus:send(Alice, muc_helper:stanza_muc_enter_room(RoomJid, AliceUsername)),
-              %% We don't care about presences from Alice, escalus would filter them out
+              wait_for_muc_presence(Alice, RoomJid, AliceUsername),
               wait_for_subject(Alice),
 
               send_n_muc_messages(Alice, RoomAddr, 3),
@@ -489,6 +487,7 @@ test_muc_conversation_history(Config0) ->
 
               wait_for_muc_presence(Eve, RoomJid, AliceUsername),
               wait_for_muc_presence(Eve, RoomJid, EveUsername),
+              wait_for_muc_presence(Alice, RoomJid, EveUsername),
 
               %% XEP-0045: After sending the presence broadcast (and only after doing so),
               %% the service MAY then send discussion history, the room subject,
@@ -659,8 +658,6 @@ test_pm_with_graceful_reconnection_to_different_server(Config) ->
               escalus_client:send(Alice, chat_with_seqnum(Eve, <<"Hi from Europe1!">>)),
 
               NewEve = connect_from_spec(EveSpec2, Config),
-              EveNode2 = ct:get_config({hosts, mim, node}),
-              wait_for_registration(NewEve, EveNode2),
 
               ok = rpc(asia_node, sys, resume, [C2sPid]),
 
@@ -742,8 +739,6 @@ do_test_pm_with_ungraceful_reconnection_to_different_server(Config0, BeforeResum
 
               %% Connect another one, we hope the message would be rerouted
               NewEve = connect_from_spec(EveSpec2, Config),
-              EveNode2 = ct:get_config({hosts, mim, node}),
-              wait_for_registration(NewEve, EveNode2),
 
               BeforeResume(),
 
@@ -1069,8 +1064,8 @@ hide_node(NodeName, Config) ->
 
 connect_from_spec(UserSpec, Config) ->
     {ok, User} = escalus_client:start(Config, UserSpec, <<"res1">>),
-    escalus_connection:set_filter_predicate(User, fun(S) -> not escalus_pred:is_presence(S) end),
     escalus_story:send_initial_presence(User),
+    escalus:assert(is_presence, escalus_client:wait_for_stanza(User)),
     User.
 
 chat_with_seqnum(To, Text) ->
@@ -1252,23 +1247,11 @@ trigger_rebalance(NodeName, DestinationDomain) when is_binary(DestinationDomain)
 %% -----------------------------------------------------------------------
 %% Escalus-related helpers
 
+%% Receive messages with Bodies in any order, skipping presences from stream resumption
 user_receives(User, Bodies) ->
-    ExpectedLength = length(Bodies),
-    Messages = escalus_client:wait_for_stanzas(User, ExpectedLength),
-    SortedMessages = order_by_seqnum(Messages),
-    case length(Messages) of
-        ExpectedLength ->
-            Checks = [escalus_pred:is_chat_message(Body, Stanza) || {Body, Stanza} <- lists:zip(Bodies, SortedMessages)],
-            case lists:all(fun(Check) -> Check end, Checks) of
-                true ->
-                    ok;
-                false ->
-                    ct:fail({user_receives_failed, {wanted, Bodies}, {received, SortedMessages}, {check, Checks}})
-            end;
-        _ ->
-            ct:fail({user_receives_not_enough, {wanted, Bodies}, {received, SortedMessages}})
-    end.
-
+    Opts = #{pred => fun(Stanza) -> not escalus_pred:is_presence(Stanza) end},
+    Checks = [fun(Stanza) -> escalus_pred:is_chat_message(Body, Stanza) end || Body <- Bodies],
+    escalus:assert_many(Checks, [escalus_connection:receive_stanza(User, Opts) || _ <- Bodies]).
 
 %% -----------------------------------------------------------------------
 %% Refreshing helpers
@@ -1306,19 +1289,6 @@ wait_for_domain(Node, Domain) ->
         end,
     mongoose_helper:wait_until(F, true, #{name => {wait_for_domain, Node, Domain}}).
 
-%% We receive presence BEFORE session is registered in ejabberd_sm.
-%% So, to ensure that we processed do_open_session completely, let's send a "ping".
-%% by calling the c2s process.
-%% That call would only return, when all messages in erlang message queue
-%% are processed.
-wait_for_registration(Client, Node) ->
-    RPCSpec = #{node => Node},
-    mongoose_helper:wait_until(fun() -> is_pid(mongoose_helper:get_session_pid(Client, RPCSpec)) end, true,
-                               #{name => wait_for_session}),
-    C2sPid = mongoose_helper:get_session_pid(Client, RPCSpec),
-    rpc:call(node(C2sPid), ejabberd_c2s, get_info, [C2sPid]),
-    ok.
-
 
 %% -----------------------------------------------------------------------
 %% Ensure, that endpoints are up

big_tests/tests/race_conditions_SUITE.erl

@@ -89,7 +89,7 @@ delayiq_handler_works(Config) ->
             {ok, DelayIqConfirmation} = receive_delayiq_confirmation(),
             %% HAVING waiting iq handler
             resume_delayiq(DelayIqConfirmation),
-            %% HAVING result IQ stanza put into ejabberd_c2s process
+            %% HAVING result IQ stanza put into mongoose_c2s process
             %% (if there is any)
             wait_for_delayiq_handler_to_finish(DelayIqConfirmation),
             receive_delayiq(Alice, RequestIQ),
@@ -100,8 +100,7 @@ delayiq_handler_works(Config) ->
 %% IQ result from IQ, sent using the old connection.
 %%
 %% If old and new resources are different, that the stanza would not be routed
-%% to the new connection in any case (with or without
-%% check_incoming_accum_for_conflicts check in ejabberd_c2s).
+%% to the new connection in any case
 ignore_iq_result_from_old_session(Config) ->
     escalus:fresh_story_with_config(Config, [{alice, 1}],
         fun(FreshConfig, Alice) ->
@@ -118,7 +117,7 @@ ignore_iq_result_from_old_session(Config) ->
             Resource = escalus_client:resource(Alice),
             Alice2 = login_send_and_receive_presence(FreshConfig, alice, Resource),
             resume_delayiq(DelayIqConfirmation),
-            %% HAVING result IQ stanza put into ejabberd_c2s process
+            %% HAVING result IQ stanza put into mongoose_c2s process
             %% (if there is any)
             %% We expect, that Alice2 c2s process receives the route message,
             %% but would ignore it.

src/c2s/mongoose_c2s.erl

@@ -1039,24 +1039,24 @@ get_lang(#c2s_data{lang = Lang}) ->
 get_stream_id(#c2s_data{streamid = StreamId}) ->
     StreamId.
 
--spec get_mod_state(data(), atom()) -> {ok, term()} | {error, not_found}.
-get_mod_state(#c2s_data{state_mod = Handlers}, HandlerName) ->
-    case maps:get(HandlerName, Handlers, undefined) of
+-spec get_mod_state(data(), module()) -> {ok, term()} | {error, not_found}.
+get_mod_state(#c2s_data{state_mod = ModStates}, ModName) ->
+    case maps:get(ModName, ModStates, undefined) of
         undefined -> {error, not_found};
-        HandlerState -> {ok, HandlerState}
+        ModState -> {ok, ModState}
     end.
 
 -spec get_listener_opts(data()) -> listener_opts().
 get_listener_opts(#c2s_data{listener_opts = ListenerOpts}) ->
     ListenerOpts.
 
--spec merge_mod_state(data(), map()) -> data().
-merge_mod_state(StateData = #c2s_data{state_mod = StateHandlers}, MoreHandlers) ->
-    StateData#c2s_data{state_mod = maps:merge(StateHandlers, MoreHandlers)}.
+-spec merge_mod_state(data(), #{module() => term()}) -> data().
+merge_mod_state(StateData = #c2s_data{state_mod = ModStates}, MoreModStates) ->
+    StateData#c2s_data{state_mod = maps:merge(ModStates, MoreModStates)}.
 
--spec remove_mod_state(data(), atom()) -> data().
-remove_mod_state(StateData = #c2s_data{state_mod = Handlers}, HandlerName) ->
-    StateData#c2s_data{state_mod = maps:remove(HandlerName, Handlers)}.
+-spec remove_mod_state(data(), module()) -> data().
+remove_mod_state(StateData = #c2s_data{state_mod = ModStates}, ModName) ->
+    StateData#c2s_data{state_mod = maps:remove(ModName, ModStates)}.
 
 -spec merge_states(data(), data()) -> data().
 merge_states(S0 = #c2s_data{}, S1 = #c2s_data{}) ->

src/c2s/mongoose_c2s_acc.erl

@@ -141,8 +141,8 @@ to_acc_many(Acc, C2SAcc, [Pair | Pairs]) ->
     to_acc_many(Acc, NewCAcc, Pairs).
 
 -spec to_c2s_acc(mongoose_c2s_acc:t(), pair()) -> mongoose_c2s_acc:t().
-to_c2s_acc(C2SAcc = #{state_mod := Handlers}, {state_mod, {Name, Handler}}) ->
-    C2SAcc#{state_mod := Handlers#{Name => Handler}};
+to_c2s_acc(C2SAcc = #{state_mod := ModStates}, {state_mod, {ModName, ModState}}) ->
+    C2SAcc#{state_mod := ModStates#{ModName => ModState}};
 to_c2s_acc(C2SAcc = #{actions := Actions}, {actions, NewActions}) when is_list(NewActions) ->
     C2SAcc#{actions := lists:reverse(NewActions) ++ Actions};
 to_c2s_acc(C2SAcc = #{actions := Actions}, {actions, Action}) ->

src/ejabberd_sm.erl

@@ -735,7 +735,6 @@ is_privacy_allow(From, To, Acc, Packet) ->
 
 
 %% @doc Check if privacy rules allow this delivery
-%% Function copied from ejabberd_c2s.erl
 -spec is_privacy_allow(From, To, Acc, Packet, PrivacyList) -> boolean() when
       From :: jid:jid(),
       To :: jid:jid(),