anchor.diff:          {nodelay, true}].
   
   %%------------------------------------------------------------------------------------------------
  -%% User presence
  -%%------------------------------------------------------------------------------------------------
  --spec send_presence_available(escalus:client()) -> ok.
  -send_presence_available(Client) ->
  -    Pres = escalus_stanza:presence(<<"available">>),
  -    escalus:send_and_wait(Client, Pres).
  +%% Node creation
  +%%------------------------------------------------------------------------------------------------
  +create_pubsub_node(Client) ->
  +    ReqId = ?NODE_CREATION_ID,
  +    Request = publish_pubsub_stanza(Client, ReqId, #xmlel{name = <<"nothing">>}),
  +    escalus:send(Client, Request),
  +
  +    {CreateNodeTime, CreateNodeResult} = timer:tc(
  +        fun() ->
  +            catch escalus:wait_for_stanza(Client, get_parameter(iq_timeout))
  +        end),
  +
  +    case {escalus_pred:is_iq_result(Request, CreateNodeResult), CreateNodeResult} of
  +        {true, _} ->
  +%%            lager:debug("node creation ~p (~p)", [?NODE, self()]),
  +            amoc_metrics:update_counter(node_creation_success),
  +            amoc_metrics:update_time(node_creation, CreateNodeTime);
  +        {false, {'EXIT', {timeout_when_waiting_for_stanza, _}}} ->
  +            amoc_metrics:update_counter(node_creation_timeout),
  +            lager:error("Timeout creating node: ~p", [CreateNodeResult]),
  +            exit(node_creation_timeout);
  +        {false, _} ->
  +            amoc_metrics:update_counter(node_creation_failure),
  +            lager:error("Error creating node: ~p", [CreateNodeResult]),
  +            exit(node_creation_failed)
  +    end.
  +
  +%%------------------------------------------------------------------------------------------------
  +%% User presence & caps
  +%%------------------------------------------------------------------------------------------------
  +send_presence(From, Type, To = #client{}) ->
  +    ToJid = escalus_client:short_jid(To),
  +    send_presence(From, Type, ToJid);
  +send_presence(From, Type, To) ->
  +    Presence = escalus_stanza:presence_direct(To, Type),
  +    escalus_client:send(From, Presence).
  +
  +send_presence_with_caps(Client) ->
  +    lager:debug("Send presence with caps ~p, (~p).", [escalus_client:short_jid(Client), self()]),
  +    Presence = escalus_stanza:presence(<<"available">>, [caps()]),
  +    escalus:send(Client, Presence).
  +
  +caps() ->
  +    #xmlel{name = <<"c">>,
  +        attrs = [{<<"xmlns">>, <<"http://jabber.org/protocol/caps">>},
  +            {<<"hash">>, <<"sha-1">>},
  +            {<<"node">>, <<"http://www.chatopus.com">>},
  +            {<<"ver">>, ?CAPS_HASH}]}.
   
   %%------------------------------------------------------------------------------------------------
   %% Room creation
   %%------------------------------------------------------------------------------------------------
  -create_muc_light_room(Client) ->
  +request_muc_light_room(Client) ->
       amoc_throttle:send_and_wait(?ROOM_CREATION_THROTTLING, create_room),
  -
  +    Id = ?ROOM_CREATION_ID,
       MucHost = get_parameter(muc_host),
  -    CreateRoomStanza = escalus_stanza:iq_set(<<"urn:xmpp:muclight:0#create">>, []),
  +    CreateRoomStanza = escalus_stanza:iq_set(?NS_MUC_LIGHT_CREATION, []),
       CreateRoomStanzaWithTo = escalus_stanza:to(CreateRoomStanza, MucHost),
  -    lager:debug("Create stanza: ~p~n", [CreateRoomStanzaWithTo]),
  -
  -    {CreateRoomTime, AffMessage} = timer:tc(fun() -> escalus:send_and_wait(Client, CreateRoomStanzaWithTo) end),
  -    amoc_metrics:update_time(room_creation, CreateRoomTime),
  -
  -    true = has_same_id(AffMessage, CreateRoomStanzaWithTo),
  -
  -    RoomJid = exml_query:attr(AffMessage, <<"from">>),
  -    erlang:put(room_jid,RoomJid),
  -    add_room(RoomJid),
  -
  -    IqResponse = escalus:wait_for_stanza(Client),
  -    lager:debug("Create room iq response: ~p~n", [IqResponse]),
  -    escalus:assert(is_iq_result, [CreateRoomStanza], IqResponse),
  -
  -    amoc_metrics:update_counter(room, 1).
  +    CreateRoomStanzaWithId = escalus_stanza:set_id(CreateRoomStanzaWithTo, Id),
  +
  +    escalus:send(Client, CreateRoomStanzaWithId),
  +    {os:system_time(microsecond), Id}.
   
   %%------------------------------------------------------------------------------------------------
   %% Room affiliation change
   %%------------------------------------------------------------------------------------------------
   add_users_to_room(Client, Jids) ->
  -    RoomJid = erlang:get(room_jid),
  -    erlang:put(room_jid, RoomJid),
  +    Id = iq_id(affiliation, Client),
  +    RoomJid = erlang:get(my_room),
       AffList = [#xmlel{name = <<"user">>,
           attrs = [{<<"affiliation">>, <<"member">>}],
           children = [#xmlcdata{content = Jid}]} || Jid <- Jids],
  -    AffChangeStanza = escalus_stanza:iq_set(<<"urn:xmpp:muclight:0#affiliations">>, AffList),
  -    lager:debug("Adding users: ~p ~p", [RoomJid,Jids]),
  -    escalus:send(Client, escalus_stanza:to(AffChangeStanza, RoomJid)).
  -
  -%%------------------------------------------------------------------------------------------------
  -%% Sending messages
  -%%------------------------------------------------------------------------------------------------
  +    AffChangeStanza = escalus_stanza:iq_set(?NS_MUC_LIGHT_AFFILIATIONS, AffList),
  +    AffChangeStanzaWithId = escalus_stanza:set_id(AffChangeStanza, Id),
  +    lager:debug("Adding users to room: ~p", [Jids]),
  +    escalus:send(Client, escalus_stanza:to(AffChangeStanzaWithId, RoomJid)),
  +    {os:system_time(microsecond), Id}.
  +
  +%%------------------------------------------------------------------------------------------------
  +%% Sending muc_light messages
  +%%------------------------------------------------------------------------------------------------
  +send_message_to_room(Client, undefined) ->
  +    RoomJid = erlang:get(my_room),
  +    send_message_to_room(Client, RoomJid);
   send_message_to_room(Client, RoomJid) ->
  -    TimeStamp = io_lib:format("~p", [os:system_time(microsecond)]),
  -    MessageBody = #xmlel{name     = <<"body">>,
  -                         children = [#xmlcdata{content = TimeStamp}]},
  -    Message = #xmlel{name     = <<"message">>,
  -                     attrs    = [{<<"to">>, RoomJid},
  -                                 {<<"type">>, <<"groupchat">>}],
  -                     children = [MessageBody]},
  +    PayloadSize = get_parameter(publication_size),
  +    MessageBody = item_content(PayloadSize),
  +    Message = #xmlel{name = <<"message">>,
  +        attrs = [{<<"to">>, RoomJid},
  +            {<<"type">>, <<"groupchat">>}],
  +        children = [MessageBody]},
       escalus:send(Client, Message).
   
   %%------------------------------------------------------------------------------------------------
  -%% Stanza helpers
  -%%------------------------------------------------------------------------------------------------
  --spec has_same_id(exml:element(), exml:element()) -> boolean().
  -has_same_id(OrigStanza, Stanza) ->
  -    OrigId = exml_query:attr(OrigStanza, <<"id">>),
  -    Id = exml_query:attr(Stanza, <<"id">>),
  -    OrigId =:= Id.
  -
  --spec get_muc_timestamp(exml:element()) -> integer().
  -get_muc_timestamp(Stanza) ->
  -    TimeStampBin = exml_query:path(Stanza, [{element, <<"body">>}, cdata]),
  -    binary_to_integer(TimeStampBin).
  -
  -
  -%%------------------------------------------------------------------------------------------------
  -%% rooms management
  -%%------------------------------------------------------------------------------------------------
  -add_room(RoomJid) ->
  +%% Item publishing
  +%%------------------------------------------------------------------------------------------------
  +
  +publish_pubsub_item(Client) ->
  +    Id = iq_id(publish, Client),
  +    PayloadSize = get_parameter(publication_size),
  +    Content = item_content(PayloadSize),
  +    Request = publish_pubsub_stanza(Client, Id, Content),
  +    escalus:send(Client, Request),
  +    {os:system_time(microsecond), Id}.
  +
  +publish_pubsub_stanza(Client, Id, Content) ->
  +    ItemId = <<"current">>,
  +    escalus_pubsub_stanza:publish(Client, ItemId, Content, Id, ?NODE).
  +
  +item_content(PayloadSize) ->
  +    Payload = #xmlcdata{content = <<<<"A">> || _ <- lists:seq(1, PayloadSize)>>},
  +    #xmlel{
  +        name = <<"entry">>,
  +        attrs = [{<<"timestamp">>, integer_to_binary(os:system_time(microsecond))},
  +            {<<"jid">>, erlang:get(jid)}],
  +        children = [Payload]}.
  +
  +%%------------------------------------------------------------------------------------------------
  +%% Item processing
  +%%------------------------------------------------------------------------------------------------
  +
  +process_message(Stanza, RecvTimeStamp) ->
  +    Type = exml_query:attr(Stanza, <<"type">>),
  +    case Type of
  +        <<"groupchat">> -> process_muc_light_message(Stanza, RecvTimeStamp);
  +        _ -> process_pubsub_msg(Stanza, RecvTimeStamp)
  +    end.
  +
  +process_pubsub_msg(#xmlel{name = <<"message">>} = Stanza, TS) ->
  +    Entry = exml_query:path(Stanza, [{element, <<"event">>}, {element, <<"items">>},
  +        {element, <<"item">>}, {element, <<"entry">>}]),
  +    case Entry of
  +        undefined -> ok;
  +        _ ->
  +            case {exml_query:attr(Entry, <<"jid">>), erlang:get(jid)} of
  +                {JID, JID} -> schedule_node_publishing(self());
  +                _ -> ok
  +            end,
  +            TimeStampBin = exml_query:attr(Entry, <<"timestamp">>),
  +            TimeStamp = binary_to_integer(TimeStampBin),
  +            TTD = TS - TimeStamp,
  +%%            lager:debug("pubsub time to delivery ~p", [TTD]),
  +            amoc_metrics:update_counter(pubsub_message),
  +            amoc_metrics:update_time(pubsub_message_ttd, TTD)
  +    end.
  +
  +process_muc_light_message(Stanza, RecvTimeStamp) ->
  +    case exml_query:subelement(Stanza, <<"x">>) of
  +        undefined ->
  +            handle_normal_muc_light_message(Stanza, RecvTimeStamp);
  +        #xmlel{name = <<"x">>, attrs = [{<<"xmlns">>, ?NS_MUC_LIGHT_AFFILIATIONS}], children = _} ->
  +            handle_muc_light_affiliation_message(Stanza),
  +            amoc_metrics:update_counter(muc_light_affiliation_change_messages);
  +        _ -> lager:error("Unknown message.")
  +    end.
  +
  +handle_normal_muc_light_message(Stanza, RecvTimeStamp) ->
  +    ReqTimeStampBin = exml_query:path(Stanza, [{element, <<"entry">>}, {attr, <<"timestamp">>}]),
  +    ReqTimeStamp = binary_to_integer(ReqTimeStampBin),
  +
  +    RoomBareJid = get_sender_bare_jid(Stanza),
  +
  +    From = exml_query:path(Stanza, [{element, <<"entry">>}, {attr, <<"jid">>}]),
  +    case erlang:get(jid) of
  +        From -> schedule_room_publishing(self(), RoomBareJid);
  +        _ -> ok
  +    end,
  +
  +    TTD = RecvTimeStamp - ReqTimeStamp,
  +%%    lager:debug("muc light time to delivery ~p", [TTD]),
  +    amoc_metrics:update_counter(muc_light_message),
  +    amoc_metrics:update_time(muc_light_ttd, TTD).
  +
  +handle_muc_light_affiliation_message(Stanza) ->
  +    amoc_metrics:update_counter(muc_light_affiliation_change_messages),
  +    case exml_query:subelement(Stanza, <<"prev-version">>) of
  +        % actually XEP states only that prev-version SHOULD NOT be sent to new users - not sure if can rely on that
  +        undefined -> handle_first_affiliation_message(Stanza);
  +        _ -> ok % drop affiliation change stanzas
  +    end.
  +
  +handle_first_affiliation_message(Stanza) ->
  +    RoomJid = exml_query:attr(Stanza, <<"from">>),
       case erlang:get(rooms) of
           undefined ->
  -            erlang:put(rooms, [RoomJid]);
  -        Rooms ->
  -            case lists:member(RoomJid, Rooms) of
  +            erlang:put(rooms, [RoomJid]),
  +            erlang:put(my_room, RoomJid);
  +        RoomList ->
  +            case lists:member(RoomJid, RoomList) of
                   true -> ok;
  -                _ -> erlang:put(rooms, [RoomJid | Rooms])
  +                false -> erlang:put(rooms, [RoomJid | RoomList])
               end
       end.
   
  -get_room() ->
  -    case erlang:get(rooms) of
  -        undefined -> error(no_rooms);
  -        Rooms ->
  -            Size = length(Rooms),
  -            N = random:uniform(Size),
  -            lists:nth(N, Rooms)
  -    end.
  +process_presence(Client, Stanza) ->
  +    case exml_query:attr(Stanza, <<"type">>) of
  +        <<"subscribe">> ->
  +            From = exml_query:attr(Stanza, <<"from">>),
  +            send_presence(Client, <<"subscribed">>, From);
  +        _ ->
  +            ok %%it's ok to just ignore other presence notifications
  +    end.
  +
  +process_iq(Client, #xmlel{name = <<"iq">>} = Stanza, TS, Requests) ->
  +    Id = exml_query:attr(Stanza, <<"id">>),
  +    Type = exml_query:attr(Stanza, <<"type">>),
  +    NS = exml_query:path(Stanza, [{element, <<"query">>}, {attr, <<"xmlns">>}]),
  +    case {Type, NS, Id, maps:get(Id, Requests, undefined)} of
  +        {<<"result">>, undefined, ?ROOM_CREATION_ID, {Tag, ReqTS}} ->
  +            handle_muc_light_room_iq_result(Stanza, {Tag, TS - ReqTS}),
  +            send_info_to_coordinator(Client);
  +        {<<"result">>, _, <<"affiliation", _/binary>>, {Tag, ReqTS}} ->
  +            handle_affiliation_change_iq(Stanza, {Tag, TS - ReqTS});
  +        {<<"get">>, ?NS_DISCO_INFO, _, undefined} ->
  +            handle_disco_query(Client, Stanza);
  +        {<<"set">>, ?NS_ROSTER, _, undefined} ->
  +            ok; %%it's ok to just ignore roster pushes
  +        {_, undefined, <<"publish", _/binary>>, undefined} ->
  +            lager:warning("unknown publish iq ~p", [Stanza]);
  +        {_, undefined, <<"publish", _/binary>>, {Tag, ReqTS}} ->
  +            handle_publish_resp(Stanza, {Tag, TS - ReqTS});
  +        _ ->
  +            lager:warning("unexpected iq ~p", [Stanza])
  +    end,
  +    maps:remove(Id, Requests).
  +
  +handle_muc_light_room_iq_result(CreateRoomResult, {Tag, RoomCreationTime}) ->
  +    case {escalus_pred:is_iq_result(CreateRoomResult), CreateRoomResult} of
  +        {true, _} ->
  +            lager:debug("Room creation ~p took ~p", [self(), RoomCreationTime]),
  +            amoc_metrics:update_time(room_creation, RoomCreationTime),
  +            IqTimeout = get_parameter(iq_timeout),
  +            case Tag of
  +                new when IqTimeout * 1000 > RoomCreationTime ->
  +                    amoc_metrics:update_counter(room_creation_success);
  +                new ->
  +                    amoc_metrics:update_counter(room_creation_timeout);
  +                timeout -> ok %% do nothing, it's already reported as timeout
  +            end;
  +        {false, _} ->
  +            amoc_metrics:update_counter(room_creation_failure),
  +            lager:error("Error creating room: ~p", [CreateRoomResult]),
  +            exit(room_creation_failed)
  +    end.
  +
  +send_info_to_coordinator(Client) ->
  +    lager:debug("Process ~p, sending info about myself to coordinator", [self()]),
  +    coordinator:register_user(Client).
  +
  +handle_affiliation_change_iq(AffiliationChangeResult, {Tag, AffiliationChangeTime}) ->
  +    case {escalus_pred:is_iq_result(AffiliationChangeResult), AffiliationChangeResult} of
  +        {true, _} ->
  +            lager:debug("Room creation ~p took ~p", [self(), AffiliationChangeTime]),
  +            amoc_metrics:update_time(room_affiliation_change, AffiliationChangeTime),
  +            IqTimeout = get_parameter(iq_timeout),
  +            case Tag of
  +                new when IqTimeout * 1000 > AffiliationChangeTime ->
  +                    amoc_metrics:update_counter(room_affiliation_change_success);
  +                new ->
  +                    amoc_metrics:update_counter(room_affiliation_change_timeout);
  +                timeout -> ok %% do nothing, it's already reported as timeout
  +            end;
  +        {false, _} ->
  +            amoc_metrics:update_counter(room_affiliation_change_failure),
  +            lager:error("Error affiliation change: ~p", [AffiliationChangeTime]),
  +            exit(affiliation_change_timeout)
  +    end.
  +
  +handle_publish_resp(PublishResult, {Tag, PublishTime}) ->
  +    IqTimeout = get_parameter(iq_timeout),
  +    case escalus_pred:is_iq_result(PublishResult) of
  +        true ->
  +            amoc_metrics:update_counter(publication_result),
  +            amoc_metrics:update_time(pubsub_publication, PublishTime),
  +            case Tag of
  +                new when IqTimeout * 1000 > PublishTime ->
  +                    amoc_metrics:update_counter(publication_success);
  +                new ->
  +                    amoc_metrics:update_counter(publication_timeout);
  +                timeout -> ok %% do nothing, it's already reported as timeout
  +            end;
  +        _ ->
  +            amoc_metrics:update_counter(publication_error),
  +            lager:error("Error publishing failed: ~p", [PublishResult]),
  +            exit(publication_failed)
  +    end.
  +
  +
  +handle_disco_query(Client, DiscoRequest) ->
  +    lager:debug("handle_disco_query ~p", [self()]),
  +    QueryEl = escalus_stanza:query_el(<<"http://jabber.org/protocol/disco#info">>,
  +        feature_elems()),
  +    DiscoResult = escalus_stanza:iq_result(DiscoRequest, [QueryEl]),
  +    escalus:send(Client, DiscoResult).
  +
  +feature_elems() ->
  +    NodeNs = ?PEP_NODE_NS,
  +    [#xmlel{name = <<"identity">>,
  +        attrs = [{<<"category">>, <<"client">>},
  +            {<<"name">>, <<"Psi">>},
  +            {<<"type">>, <<"pc">>}]},
  +        #xmlel{name = <<"feature">>,
  +            attrs = [{<<"var">>, <<"http://jabber.org/protocol/disco#info">>}]},
  +        #xmlel{name = <<"feature">>,
  +            attrs = [{<<"var">>, NodeNs}]},
  +        #xmlel{name = <<"feature">>,
  +            attrs = [{<<"var">>, <<NodeNs/bitstring, "+notify">>}]}].
  +
  +%%------------------------------------------------------------------------------------------------
  +%% Stanza helpers
  +%%------------------------------------------------------------------------------------------------
  +
  +iq_id(Type, Client) ->
  +    UserName = escalus_utils:get_username(Client),
  +    Suffix = random_suffix(),
  +    list_to_binary(io_lib:format("~s-~s-~p",
  +        [Type, UserName, Suffix])).
  +
  +random_suffix() ->
  +    Suffix = base64:encode(crypto:strong_rand_bytes(5)),
  +    re:replace(Suffix, "/", "_", [global, {return, binary}]).
  +
  +get_sender_bare_jid(Stanza) ->
  +    From = exml_query:attr(Stanza, <<"from">>),
  +    [BareJid | _] = binary:split(From, <<"/">>),
  +    BareJid.
  +
   %%------------------------------------------------------------------------------------------------
   %% Config helpers
   %%------------------------------------------------------------------------------------------------
  
positive.code: -include_lib("escalus/include/escalus.hrl").
  -include_lib("escalus/include/escalus_xmlns.hrl").
  
  -required_variable({'ROOM_SIZE', <<"Number of users in a room.">>}).
  -required_variable({'ROOM_CREATION_RATE', <<"Rate of room creations (per minute, def:600)">>}).
  -required_variable({'GDPR_REMOVAL_RATE', <<"Rate of user removals (per minute, def:1)">>}).
  -required_variable({'PUBLICATION_RATE', <<"Rate of publications (per minute, def:1500)">>}).
  -required_variable({'ACTIVATION_POLICY', <<"Publish after subscribtion of (def: all_rooms | n_rooms)"/utf8>>}).
  -required_variable({'MIM_HOST', <<"The virtual host served by the server (def: <<\"localhost\">>)"/utf8>>}).
  -required_variable({'MUC_HOST', <<"The virtual MUC host served by the server (def: <<\"muclight.localhost\">>)"/utf8>>}).
  -required_variable({'MIM_REST_ENDPOINT', <<"MongooseIM REST endpoint (def: \"http://localhost:8088\">)">>}).
  
  -define(ALL_PARAMETERS, [
      {room_size, 'ROOM_SIZE', 5, positive_integer},
      {room_creation_rate, 'ROOM_CREATION_RATE', 600, positive_integer},
      {gdpr_removal_rate, 'GDPR_REMOVAL_RATE', 1, positive_integer},
      {publication_rate, 'PUBLICATION_RATE', 1500, positive_integer},
      {activation_policy, 'ACTIVATION_POLICY', n_rooms, [all_rooms, n_rooms]},
      {mim_host, 'MIM_HOST', <<"localhost">>, bitstring},
      {muc_host, 'MUC_HOST', <<"muclight.localhost">>, bitstring},
      {mim_rest_endpoint, 'MIM_REST_ENDPOINT', "http://localhost:8088", none}
  ]).
  
  -define(ROOM_CREATION_THROTTLING, room_creation).
  -define(PUBLICATION_THROTTLING, publication).
  -define(REMOVAL_THROTTLING, user_removal).
  
  %% this prevents user processes from restarting
  -define(STOP_USER, throw(stop)).
  
  -export([init/0, start/2]).
  
  
negative.code:          OwnerPid ! {add_users, MemberJids}
       end || {OwnerPid, OwnerJid} <- PidsAndJids].
  
  schedule_removal(Pid) ->
      amoc_throttle:send(?REMOVAL_THROTTLING, Pid, remove_user).
  
  %%------------------------------------------------------------------------------------------------
  %% User
  