anchor.diff:          _ -> %% [] or {error, {no_such_group, ?GROUP_NAME}}
               timer:sleep(100),
               get_coordinator_pid()
  -    end.
  -
  -%%------------------------------------------------------------------------------------------------
  -%% Helper
  -%%------------------------------------------------------------------------------------------------
  -
  -start_helper() ->
  -    Self = self(),
  -    lager:debug("helper process ~p", [Self]),
  -    get_coordinator_pid(), %% ensure than coordinator is created
  -    pg2:join(?HELPERS_GROUP_NAME, Self),
  -    helper_loop().
  -
  -helper_loop() ->
  -    UserPid = receive
  -                  {allocate_helper, Pid} ->
  -                      Pid
  -              end,
  -    UserPid ! allocated,
  -    receive
  -        release_helper ->
  -            helper_loop()
  -    end.
  -
  -
  -get_helper() ->
  -    case pg2:get_members(?HELPERS_GROUP_NAME) of
  -        [_ | _] = List -> %% nonempty list
  -            N = rand:uniform(length(List)),
  -            lists:nth(N, List);
  -        _ -> %% [] or {error, {no_such_group, ?GROUP_NAME}}
  -            timer:sleep(100),
  -            get_helper()
  -    end.
  -
  -create_pubsub_node_for_user(Client) ->
  -    %% here helper processes are used as just pool of mutexes.
  -    %% we can send message from the helper process using Client.
  -    %% however we cannot receive any response back. triggering
  -    %% escalus:wait_for_stanza/2 from the helper process always
  -    %% results in timeout. as incoming message with stanza arrives
  -    %% to the Client's creator/owner (user) process, not the
  -    %% function caller process.
  -    Helper = get_helper(),
  -    Helper ! {allocate_helper, self()},
  -    receive
  -        allocated ->
  -            create_pubsub_node(Client),
  -            Helper ! release_helper
       end.
   
   %%------------------------------------------------------------------------------------------------
  
positive.code: update_timeout_metrics(Id) ->
      lager:error("unknown iq id ~p", Id).
  
  schedule_publishing(SchedulingPolicy) ->
      case get_parameter(scheduling_policy) of
          SchedulingPolicy ->
              DelayBetweenMessages = get_parameter(publication_delay),
              erlang:send_after(DelayBetweenMessages, self(), publish_item);
          _ -> ok
      end.
  
  
  %%------------------------------------------------------------------------------------------------
  %% User connection
  
negative.code: 
  -required_variable({'IQ_TIMEOUT',         <<"IQ timeout (milliseconds, def: 10000ms)"/utf8>>}).
  -reqeired_variable({'COORDINATOR_DELAY',  <<"Delay after N subscriptions (milliseconds, def: 0ms)"/utf8>>}).
  -required_variable({'NODE_CREATION_RATE', <<"Rate of node creations (per minute, def:600)">>}).
  -required_variable({'PUBLICATION_SIZE',   <<"Size of additional payload (bytes, def:300)">>}).
  -required_variable({'PUBLICATION_RATE',   <<"Rate of publications (per minute, def:1500)">>}).
  -required_variable({'N_OF_SUBSCRIBERS',   <<"Number of subscriptions for each node (def: 50)"/utf8>>}).
  -required_variable({'ACTIVATION_POLICY',  <<"Publish after subscribtion of (def: all_nodes | n_nodes)"/utf8>>}).
  -required_variable({'MIM_HOST',           <<"The virtual host served by the server (def: <<\"localhost\">>)"/utf8>>}).
  
  -define(ALL_PARAMETERS,[
      {iq_timeout,         'IQ_TIMEOUT',                    10000, positive_integer},
      {coordinator_delay,  'COORDINATOR_DELAY',                 0, nonnegative_integer},
      {node_creation_rate, 'NODE_CREATION_RATE',              600, positive_integer},
      {publication_size,   'PUBLICATION_SIZE',                300, nonnegative_integer},
      {publication_rate,   'PUBLICATION_RATE',               1500, positive_integer},
      {n_of_subscribers,   'N_OF_SUBSCRIBERS',                 50, nonnegative_integer},
      {activation_policy,  'ACTIVATION_POLICY',         all_nodes, [all_nodes, n_nodes]},
      {mim_host,           'MIM_HOST',            <<"localhost">>, bitstring}
  ]).
  
  