diff --git a/README.md b/README.md index 4469327..8ab2bd1 100644 --- a/README.md +++ b/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 diff --git a/src/nrte.erl b/src/nrte.erl index dc8debf..f7bcbe4 100644 --- a/src/nrte.erl +++ b/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) -> diff --git a/src/nrte_ebus_handler.erl b/src/nrte_ebus_handler.erl index 7335532..9564cd6 100644 --- a/src/nrte_ebus_handler.erl +++ b/src/nrte_ebus_handler.erl @@ -17,7 +17,7 @@ -export([subscribe/1]). %%% CALLBACK EXPORTS --export([handle_message/3]). +-export([handle_message/2]). %%%----------------------------------------------------------------------------- %%% EXTERNAL EXPORTS @@ -25,7 +25,7 @@ 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]) diff --git a/test/nrte_eventsource_SUITE.erl b/test/nrte_eventsource_SUITE.erl index 71789ed..c93ce37 100644 --- a/test/nrte_eventsource_SUITE.erl +++ b/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). diff --git a/test/nrte_rest_SUITE.erl b/test/nrte_rest_SUITE.erl index 1cf4990..7863f1b 100644 --- a/test/nrte_rest_SUITE.erl +++ b/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,12 +156,12 @@ 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). @@ -169,12 +169,12 @@ 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 = <>, + receive + {nrte_message, ExpectedMessage} -> ok + after 1000 -> throw(timeout) + end. diff --git a/test/nrte_websocket_SUITE.erl b/test/nrte_websocket_SUITE.erl index 069d080..2b0607b 100644 --- a/test/nrte_websocket_SUITE.erl +++ b/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(<>, <<"message">>), {ws, {text, <>}} = gun:await(Pid, StreamRef), ok = gun:close(Pid).