feat(#11): Allow publishing and subscribing from Erlang through nrte (#12)

Closes #11.

Author: Alberdi

README.md

@@ -5,7 +5,6 @@ An OTP application to send and receive real time events over HTTP connections.
 ## Status
 [![GitHub branch checks state](https://github.com/nomasystems/nrte/actions/workflows/ci.yml/badge.svg)](https://github.com/nomasystems/nrte/actions/workflows/ci.yml)
 
-
 ## Prerequisites
 
 ![Min. OTP version](https://img.shields.io/badge/min._OTP-25.3.2-blue)
@@ -81,7 +80,17 @@ nrte_auth(_Headers) ->
 
 ## Bypassing the HTTP connections
 
-All events are published and received through the [`erlbus`](https://github.com/cabol/erlbus) library. It is possible to directly use it along the HTTP interfaces by publishing or subscribing to the same topics.
+Messages can also be published via `nrte:publish/2` and subscribed from `nrte:subscribe/1`. In the latter case, the caller will receive messages in the form of `{nrte_message, Data}`:
+```erl
+> nrte:subscribe([<<"example">>]).
+ok
+
+> nrte:publish(<<"example">>, <<"my-message">>).
+ok
+
+> receive {nrte_message, Data} -> Data end.
+<<"example;my-message">>
+```
 
 ## Support
 

src/nrte.erl

@@ -17,6 +17,9 @@
 %%% APPLICATION EXPORTS
 -export([start/2, stop/1]).
 
+%%% EXTERNAL EXPORTS
+-export([publish/2, subscribe/1]).
+
 %%%-----------------------------------------------------------------------------
 %%% APPLICATION EXPORTS
 %%%-----------------------------------------------------------------------------
@@ -40,3 +43,22 @@ start(_, _) ->
 stop(_) ->
     cowboy:stop_listener(nrte_listener),
     ok.
+
+%%%-----------------------------------------------------------------------------
+%%% EXTERNAL EXPORTS
+%%%-----------------------------------------------------------------------------
+-spec publish(binary(), binary()) -> ok.
+publish(Topic, Message) ->
+    ExpandedTopics = expand_topic(Topic),
+    lists:foreach(fun(T) -> ebus:pub(T, Message) end, ExpandedTopics).
+
+-spec subscribe([iodata()]) -> ok.
+subscribe(TopicList) ->
+    nrte_ebus_handler:subscribe(TopicList).
+
+%%%-----------------------------------------------------------------------------
+%%% INTERNAL FUNCTIONS
+%%%-----------------------------------------------------------------------------
+expand_topic(Topic) ->
+    Subtopics = [binary:part(Topic, {0, Pos}) || {Pos, _} <- binary:matches(Topic, <<":">>)],
+    [Topic | Subtopics].

src/nrte_ebus_handler.erl

@@ -44,4 +44,4 @@ handle_message(Message, Topic, Pid) ->
         Template,
         Replacements
     ),
-    Pid ! {ebus_message, Data}.
+    Pid ! {nrte_message, Data}.

src/nrte_eventsource.erl

@@ -26,13 +26,13 @@
 init(Req, Opts) ->
     case nrte_auth:authorization(Req, subscribe) of
         {authorized, TopicList} ->
-            nrte_ebus_handler:subscribe(TopicList),
+            nrte:subscribe(TopicList),
             Req2 = cowboy_req:stream_reply(200, #{<<"content-type">> => ?CONTENT_TYPE}, Req),
             {cowboy_loop, Req2, Opts};
         {unauthorized, Req2} ->
             {stop, Req2, Opts}
     end.
 
-info({ebus_message, Data}, Req, Opts) ->
+info({nrte_message, Data}, Req, Opts) ->
     ok = cowboy_req:stream_events(#{data => Data}, nofin, Req),
     {ok, Req, Opts}.

src/nrte_rest.erl

@@ -31,16 +31,11 @@ init(Req, Opts) ->
 %%%-----------------------------------------------------------------------------
 %%% INTERNAL FUNCTIONS
 %%%-----------------------------------------------------------------------------
-expand_topic(Topic) ->
-    Subtopics = [binary:part(Topic, {0, Pos}) || {Pos, _} <- binary:matches(Topic, <<":">>)],
-    [Topic | Subtopics].
-
 init_authorized(#{path := <<"/message">>} = Req, Opts, [<<>>]) ->
     {stop, cowboy_req:reply(400, #{}, <<"Topics required">>, Req), Opts};
 init_authorized(#{path := <<"/message">>} = Req, Opts, TopicList) ->
-    ExpandedTopics = lists:append([expand_topic(T) || T <- TopicList]),
     {ok, Body, Req2} = cowboy_req:read_body(Req),
-    lists:foreach(fun(T) -> ebus:pub(T, Body) end, ExpandedTopics),
+    lists:foreach(fun(T) -> nrte:publish(T, Body) end, TopicList),
     {ok, cowboy_req:reply(200, #{}, <<>>, Req2), Opts};
 init_authorized(Req, Opts, _TopicList) ->
     {stop, cowboy_req:reply(404, #{}, <<>>, Req), Opts}.

src/nrte_websocket.erl

@@ -23,7 +23,7 @@
 init(Req, Opts) ->
     case nrte_auth:authorization(Req, subscribe) of
         {authorized, TopicList} ->
-            nrte_ebus_handler:subscribe(TopicList),
+            nrte:subscribe(TopicList),
             {cowboy_websocket, Req, []};
         {unauthorized, Req2} ->
             {stop, Req2, Opts}
@@ -32,7 +32,7 @@ init(Req, Opts) ->
 websocket_handle(_Data, State) ->
     {ok, State}.
 
-websocket_info({ebus_message, Data}, State) ->
+websocket_info({nrte_message, Data}, State) ->
     {reply, {text, Data}, State};
 websocket_info(_Info, State) ->
     {ok, State}.