Add a PEP scenario

rebar.config

@@ -6,7 +6,7 @@
 
 { deps, [
     {amoc_arsenal, {git, "https://github.com/esl/amoc-arsenal.git", {branch, "main"}}},
-    {escalus, "~> 4.5"},
+    {escalus, "~> 4.7"},
     {exml, "~> 4.1", {pkg, hexml}},
     {gun, "~> 2.2"},
     {fusco, "~> 0.1"}

rebar.lock

@@ -12,7 +12,7 @@
  {<<"bbmustache">>,{pkg,<<"bbmustache">>,<<"1.12.2">>},1},
  {<<"cowboy">>,{pkg,<<"cowboy">>,<<"2.12.0">>},2},
  {<<"cowlib">>,{pkg,<<"cowlib">>,<<"2.15.0">>},1},
- {<<"escalus">>,{pkg,<<"escalus">>,<<"4.5.2">>},0},
+ {<<"escalus">>,{pkg,<<"escalus">>,<<"4.7.2">>},0},
  {<<"exml">>,{pkg,<<"hexml">>,<<"4.1.2">>},0},
  {<<"fast_pbkdf2">>,{pkg,<<"fast_pbkdf2">>,<<"2.0.0">>},2},
  {<<"fast_scram">>,{pkg,<<"fast_scram">>,<<"0.7.0">>},1},
@@ -36,7 +36,7 @@
  {<<"bbmustache">>, <<"0CABDCE0DB9FE6D3318131174B9F2B351328A4C0AFBEB3E6E99BB0E02E9B621D">>},
  {<<"cowboy">>, <<"F276D521A1FF88B2B9B4C54D0E753DA6C66DD7BE6C9FCA3D9418B561828A3731">>},
  {<<"cowlib">>, <<"3C97A318A933962D1C12B96AB7C1D728267D2C523C25A5B57B0F93392B6E9E25">>},
- {<<"escalus">>, <<"E25336C0C9F30A9DCB32590204D4A58357CCE127EB7D64311C514B8F97369FDB">>},
+ {<<"escalus">>, <<"B6FFA9BBAB52A98A9FC93ADEDB9061ECCE90B2E8FD146CDE611666B2A2C24CEF">>},
  {<<"exml">>, <<"4EA6B95AF18922F94FD67BBDEAA325A75CCE04487083C786E7417BA862D590E5">>},
  {<<"fast_pbkdf2">>, <<"72CDEE3C10C6B9B40E31194DE946A883CEEF6CF1F37D7FC9FD1A9D87502723F5">>},
  {<<"fast_scram">>, <<"FAD4DD185E0DEFA34B0E30654AB0CC4048E9324BCF65BB261E0A86BF3C604F9F">>},
@@ -59,7 +59,7 @@
  {<<"bbmustache">>, <<"688B33A4D5CC2D51F575ADF0B3683FC40A38314A2F150906EDCFC77F5B577B3B">>},
  {<<"cowboy">>, <<"8A7ABE6D183372CEB21CAA2709BEC928AB2B72E18A3911AA1771639BEF82651E">>},
  {<<"cowlib">>, <<"4F00C879A64B4FE7C8FCB42A4281925E9FFDB928820B03C3AD325A617E857532">>},
- {<<"escalus">>, <<"12B3C31D93C12656C04A0DE2BB945B803EE6D48F6433065E7B6860A6C8056C35">>},
+ {<<"escalus">>, <<"A48B17F2BF9BBDF801C57CB05A8045FA51B3352F6E991FD48E88AB39D1007AE1">>},
  {<<"exml">>, <<"03883AE3E27970E9DAC204B27CE330B54A6C2772455E43FE2319349DFF40B1DC">>},
  {<<"fast_pbkdf2">>, <<"74159FD09FB8BF5E97D25137C6C83C28E2CF7E97D7C127D83310DFD0904BD732">>},
  {<<"fast_scram">>, <<"39CB1D4C91C99E637E000A3864D274D8A9CB43D86EEC2495FE9546798DBFA015">>},

src/helpers/amoc_xmpp_pep.erl

@@ -0,0 +1,88 @@
+-module(amoc_xmpp_pep).
+
+-moduledoc """
+Building blocks for PEP (XEP-0163: Personal Eventing Protocol) scenarios:
+  - Determining the names of nodes to publish (see nodes_per_user)
+  - Item publishing (auto-create makes separate node creation unnecessary)
+  - Counting incoming item notifications and measuring TTD
+""".
+
+-include_lib("escalus/include/escalus_xmlns.hrl").
+-include_lib("exml/include/exml.hrl").
+-include_lib("kernel/include/logger.hrl").
+
+-export([init/0,
+         publish_item/2,
+         nodes_to_publish/0,
+         received_handler_spec/0]).
+
+-define(V(X), (fun amoc_config_validation:X/1)).
+
+-required_variable(#{name => nodes_per_user, default_value => 1,
+                     verification => ?V(nonnegative_integer),
+                     description => "Number of PEP nodes each user publishes to"}).
+
+-spec init() -> ok.
+init() ->
+    amoc_metrics:init(counters, pubsub_items_published),
+    amoc_metrics:init(counters, pubsub_notifications_received),
+    amoc_metrics:init(counters, timeouts),
+    amoc_metrics:init(times, pubsub_item_ttd),
+    amoc_metrics:init(times, response),
+    ok.
+
+-spec nodes_to_publish() -> [binary()].
+nodes_to_publish() ->
+    [node_name(NodeId) || NodeId <- lists:seq(1, cfg(nodes_per_user))].
+
+-spec received_handler_spec() -> [amoc_xmpp_handlers:handler_spec()].
+received_handler_spec() ->
+    [{fun is_pep_notification/1,
+      fun record_pep_notification/3}].
+
+-spec publish_item(escalus:client(), escalus_pubsub_stanza:pubsub_node_name()) -> ok.
+publish_item(Client, Node) ->
+    Id = escalus_stanza:id(),
+    Req = escalus_pubsub_stanza:publish(item_content(Client), Id, Node),
+    Pred = fun(Stanza) -> escalus_pred:is_iq_result(Req, Stanza) end,
+    Resp = amoc_xmpp:send_request_and_get_response(Client, Req, Pred, response, 10000),
+    ?LOG_DEBUG("~s got PEP publish response ~p", [escalus_client:username(Client), Resp]),
+    amoc_metrics:update_counter(pubsub_items_published).
+
+-spec item_content(escalus:client()) -> exml:element().
+item_content(Client) ->
+    #xmlel{
+       name = ~"entry",
+       attrs = #{~"timestamp" => integer_to_binary(os:system_time(microsecond)),
+                 ~"jid" => escalus_client:short_jid(Client)},
+       children = [#xmlcdata{content = cfg(message_body)}]}.
+
+-spec node_name(pos_integer()) -> binary().
+node_name(NodeId) ->
+    iolist_to_binary([~"test-node-", integer_to_binary(NodeId)]).
+
+-spec is_pep_notification(exml:element()) -> boolean().
+is_pep_notification(Stanza = #xmlel{name = ~"message"}) ->
+    EventXmlns = exml_query:path(Stanza, [{element, ~"event"}, {attr, ~"xmlns"}]),
+    Items = exml_query:path(Stanza, [{element, ~"event"}, {element, ~"items"}]),
+    EventXmlns =:= ?NS_PUBSUB_EVENT andalso Items =/= undefined;
+is_pep_notification(_) ->
+    false.
+
+-spec record_pep_notification(escalus:client(), exml:element(),
+                              escalus_connection:metadata()) -> ok.
+record_pep_notification(Client, Stanza, Metadata) ->
+    ?LOG_DEBUG("~s received PEP notification ~p", [escalus_client:username(Client), Stanza]),
+    amoc_metrics:update_counter(pubsub_notifications_received),
+    update_pubsub_item_ttd(Stanza, Metadata).
+
+-spec update_pubsub_item_ttd(exml:element(), escalus_connection:metadata()) -> ok.
+update_pubsub_item_ttd(Stanza, #{recv_timestamp := RecvTimestamp}) ->
+    Timestamp = exml_query:path(Stanza, [{element, ~"event"},
+                                         {element, ~"items"},
+                                         {element, ~"item"},
+                                         {element, ~"entry"},
+                                         {attr, ~"timestamp"}]),
+    amoc_metrics:update_time(pubsub_item_ttd, RecvTimestamp - binary_to_integer(Timestamp)).
+
+cfg(Name) -> amoc_config:get(Name).