feat(#19): Issue nrte_messages with the full published topic

README.md

@@ -84,11 +84,11 @@ Messages can also be published via `nrte:publish/2` and subscribed from `nrte:su
 > nrte:subscribe([<<"example">>]).
 ok
 
-> nrte:publish(<<"example">>, <<"my-message">>).
+> nrte:publish(<<"example:subexample">>, <<"my-message">>).
 ok
 
 > receive {nrte_message, Data} -> Data end.
-<<"example;my-message">>
+<<"example:subexample;my-message">>
 ```
 
 ## Subscription publications

src/nrte.erl

@@ -50,7 +50,7 @@ stop(_) ->
 -spec publish(binary(), binary()) -> ok.
 publish(Topic, Message) ->
     ExpandedTopics = expand_topic(Topic),
-    lists:foreach(fun(T) -> ebus:pub(T, Message) end, ExpandedTopics).
+    lists:foreach(fun(T) -> ebus:pub(T, {Topic, Message}) end, ExpandedTopics).
 
 -spec subscribe([iodata()]) -> ok.
 subscribe(TopicList) ->

src/nrte_ebus_handler.erl

@@ -17,15 +17,15 @@
 -export([subscribe/1]).
 
 %%% CALLBACK EXPORTS
--export([handle_message/3]).
+-export([handle_message/2]).
 
 %%%-----------------------------------------------------------------------------
 %%% EXTERNAL EXPORTS
 %%%-----------------------------------------------------------------------------
 subscribe(TopicList) ->
     lists:foreach(
         fun(Topic) ->
-            Handler = ebus_proc:spawn_handler(fun ?MODULE:handle_message/3, [Topic, self()], [link]),
+            Handler = ebus_proc:spawn_handler(fun ?MODULE:handle_message/2, [self()], [link]),
             ebus:sub(Handler, Topic)
         end,
         TopicList
@@ -34,9 +34,9 @@ subscribe(TopicList) ->
 %%%-----------------------------------------------------------------------------
 %%% CALLBACK EXPORTS
 %%%-----------------------------------------------------------------------------
-handle_message(Message, Topic, Pid) ->
+handle_message({SourceTopic, Message}, Pid) ->
     Template = nrte_conf:data_template(),
-    Replacements = [{<<"{{message}}">>, Message}, {<<"{{topic}}">>, Topic}],
+    Replacements = [{<<"{{message}}">>, Message}, {<<"{{topic}}">>, SourceTopic}],
     Data = lists:foldl(
         fun({Original, Replacement}, Acc) ->
             binary:replace(Acc, Original, Replacement, [global])

test/nrte_eventsource_SUITE.erl

@@ -118,7 +118,7 @@ receive_topic_message() ->
 
 receive_topic_message(_Conf) ->
     {ok, {Pid, StreamRef}} = connect_es("topic"),
-    ebus:pub(<<"topic">>, <<"message">>),
+    ok = nrte:publish(<<"topic">>, <<"message">>),
     {sse, #{data := [<<"topic;message">>]}} = gun:await(Pid, StreamRef, 1000),
     ok = gun:close(Pid).
 

test/nrte_rest_SUITE.erl

@@ -133,12 +133,12 @@ post_empty_body() ->
     [{userdata, [{doc, "Tests posting an empty body"}]}].
 
 post_empty_body(_Conf) ->
-    ebus:sub(self(), "topic"),
+    ok = nrte:subscribe([<<"topic">>]),
     {ok, Pid} = gun:open("localhost", 2080),
     {ok, http} = gun:await_up(Pid),
     StreamRef = gun:post(Pid, "/message?topics=topic", #{}, <<>>),
     {response, _, 200, _Headers} = gun:await(Pid, StreamRef, 1000),
-    [<<>>] = ebus_proc:messages(self()),
+    ok = nrte_receive(<<"topic">>, <<>>),
     ok = gun:close(Pid).
 
 post_no_topic() ->
@@ -156,25 +156,25 @@ post_topic_all_authorized() ->
 
 post_topic_all_authorized(_Conf) ->
     application:set_env(nrte, auth_type, {always_allow, all}),
-    ebus:sub(self(), "topic"),
+    ok = nrte:subscribe([<<"topic">>]),
     {ok, Pid} = gun:open("localhost", 2080),
     {ok, http} = gun:await_up(Pid),
     StreamRef = gun:post(Pid, "/message?topics=topic", ?HEADERS, ?TEXT),
     {response, fin, 200, _Headers} = gun:await(Pid, StreamRef, 1000),
-    [?TEXT] = ebus_proc:messages(self()),
+    ok = nrte_receive(<<"topic">>, ?TEXT),
     application:unset_env(nrte, auth_type),
     ok = gun:close(Pid).
 
 post_topic_message() ->
     [{userdata, [{doc, "Tests posting a topic message"}]}].
 
 post_topic_message(_Conf) ->
-    ebus:sub(self(), "topic"),
+    ok = nrte:subscribe([<<"topic">>]),
     {ok, Pid} = gun:open("localhost", 2080),
     {ok, http} = gun:await_up(Pid),
     StreamRef = gun:post(Pid, "/message?topics=topic", ?HEADERS, ?TEXT),
     {response, fin, 200, _Headers} = gun:await(Pid, StreamRef, 1000),
-    [?TEXT] = ebus_proc:messages(self()),
+    ok = nrte_receive(<<"topic">>, ?TEXT),
     ok = gun:close(Pid).
 
 post_topic_with_subtopics() ->
@@ -185,12 +185,12 @@ post_topic_with_subtopics(_Conf) ->
     ExpectedTopicPubs = [
         Topic, <<"topic:subtopic:subsubtopic">>, <<"topic:subtopic">>, <<"topic">>
     ],
-    ok = lists:foreach(fun(T) -> ebus:sub(self(), T) end, ExpectedTopicPubs),
+    ok = nrte:subscribe(ExpectedTopicPubs),
     {ok, Pid} = gun:open("localhost", 2080),
     {ok, http} = gun:await_up(Pid),
     StreamRef = gun:post(Pid, ["/message?topics=", Topic], ?HEADERS, ?TEXT),
     {response, fin, 200, _Headers} = gun:await(Pid, StreamRef, 1000),
-    true = length(ExpectedTopicPubs) =:= length(ebus_proc:messages(self())),
+    true = lists:all(fun(_Topic) -> nrte_receive(Topic, ?TEXT) == ok end, ExpectedTopicPubs),
     ok = gun:close(Pid).
 
 %%%-----------------------------------------------------------------------------
@@ -201,3 +201,10 @@ nrte_auth(Headers) ->
         true -> #{allowed_publications => all};
         false -> unauthorized
     end.
+
+nrte_receive(Topic, Body) ->
+    ExpectedMessage = <<Topic/binary, ";", Body/binary>>,
+    receive
+        {nrte_message, ExpectedMessage} -> ok
+    after 1000 -> throw(timeout)
+    end.

test/nrte_websocket_SUITE.erl

@@ -133,7 +133,7 @@ receive_topic_message(_Conf) ->
     ok = gun:ws_send(Pid, StreamRef, ping),
     % Wait for pong to ensure the server processed the topic subscription
     {ws, pong} = gun:await(Pid, StreamRef),
-    ebus:pub(?TOPIC, <<"message">>),
+    ok = nrte:publish(<<?TOPIC>>, <<"message">>),
     {ws, {text, <<?TOPIC, ";message">>}} = gun:await(Pid, StreamRef),
     ok = gun:close(Pid).