anchor.diff:      start_user(Client).
   
   init_metrics() ->
  -    Counters = [message_received, room, message_published],
  -    Times = [room_creation, publication_interval, message_ttd],
  +    Counters = [pubsub_message, muc_light_message,
  +        room_creation_success, room_creation_timeout, room_creation_failure,
  +        node_creation_success, node_creation_timeout, node_creation_failure,
  +        publication_query, publication_result, publication_error,
  +        publication_success, publication_timeout,
  +        muc_light_message_sent,
  +
  +        muc_light_affiliation_change_messages,
  +
  +        room_affiliation_change_success, room_affiliation_change_timeout, room_affiliation_change_failure,
  +
  +        gdpr_removal],
  +    Times = [room_creation, node_creation,
  +        pubsub_publication,
  +        pubsub_message_ttd, muc_light_ttd,
  +        room_affiliation_change,
  +        gdpr_removal],
       [amoc_metrics:init(counters, Metric) || Metric <- Counters],
       [amoc_metrics:init(times, Metric) || Metric <- Times].
   
  
positive.code:         {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).
  
  %%------------------------------------------------------------------------------------------------
  %% Room creation
  %%------------------------------------------------------------------------------------------------
  create_muc_light_room(Client) ->
      amoc_throttle:send_and_wait(?ROOM_CREATION_THROTTLING, create_room),
  
      MucHost = get_parameter(muc_host),
      CreateRoomStanza = escalus_stanza:iq_set(<<"urn:xmpp:muclight:0#create">>, []),
      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).
  
  %%------------------------------------------------------------------------------------------------
  %% Room affiliation change
  %%------------------------------------------------------------------------------------------------
  add_users_to_room(Client, Jids) ->
      RoomJid = erlang:get(room_jid),
      erlang:put(room_jid, RoomJid),
      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
  %%------------------------------------------------------------------------------------------------
  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]},
      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) ->
      case erlang:get(rooms) of
          undefined ->
              erlang:put(rooms, [RoomJid]);
          Rooms ->
              case lists:member(RoomJid, Rooms) of
                  true -> ok;
                  _ -> erlang:put(rooms, [RoomJid | Rooms])
              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.
  %%------------------------------------------------------------------------------------------------
  %% Config helpers
  %%------------------------------------------------------------------------------------------------
  
negative.code:     Plan = get_plan(),
      coordinator:start_link(Plan, Settings).
  
  node_activate_users(Pids, ActivationPolicy) ->
      case get_parameter(node_activation_policy) of
          ActivationPolicy ->
              lager:debug("Node activate users running. Policy ~p. Pids: ~p", [ActivationPolicy, Pids]),
              [schedule_node_publishing(Pid) || Pid <- Pids];
          _ -> ok
      end.
  
  room_activate_users(Pids, ActivationPolicy) ->
      case get_parameter(room_activation_policy) of
          ActivationPolicy ->
              lager:debug("Room activate users running. Policy ~p. Pids: ~p", [ActivationPolicy, Pids]),
              [schedule_room_publishing(Pid) || Pid <- Pids];
          _ -> ok
      end.
  
  activate_removal(Pids) ->
      [schedule_removal(Pid) || Pid <- Pids].
  
  make_all_clients_friends(Clients) ->
      lager:debug("Make all clients friends."),
      escalus_utils:distinct_pairs(
          fun(C1, C2) ->
              send_presence(C1, <<"subscribe">>, C2),
              send_presence(C2, <<"subscribe">>, C1)
          end, Clients).
  
  make_full_rooms(PidsAndClients) ->
      PidsAndJids = [{Pid, Client#client.jid} || {Pid, Client} <- PidsAndClients],
  