feat(#15): Allow receiving callbacks when a client connects/disconnects

README.md

@@ -91,6 +91,17 @@ ok
 <<"example;my-message">>
 ```
 
+## Subscription publications
+
+When a client connects to the `/websocket` or `/eventsource` endpoints, a special message will be published to the topic `nrte:subscription_init:{{source}}` with the client subscribed topics as message. When a client disconnects from that endpoint, a similar message will be published as `nrte:subscription_terminate:{{source}}`. It is then possible to subscribe to these special topics and receive the messages. For example:
+```erl
+{nrte_message,<<"nrte:subscription_init:websocket;topic1">>}
+{nrte_message,<<"nrte:subscription_terminate:websocket;topic1">>}
+
+{nrte_message,<<"nrte:subscription_init:eventsource;topic1;topic2;topic3">>}
+{nrte_message,<<"nrte:subscription_terminate:eventsource;topic1;topic2;topic3">>}
+```
+
 ## Support
 
 Any doubt or suggestion? Please, check out [our issue tracker](https://github.com/nomasystems/nrte/issues).

src/nrte_eventsource.erl

@@ -27,6 +27,7 @@ init(Req, Opts) ->
     case nrte_auth:authorization(Req, subscribe) of
         {authorized, TopicList} ->
             nrte:subscribe(TopicList),
+            nrte_publications:subscription_init_link_terminate(<<"eventsource">>, TopicList),
             Req2 = cowboy_req:stream_reply(200, #{<<"content-type">> => ?CONTENT_TYPE}, Req),
             {cowboy_loop, Req2, Opts};
         {unauthorized, Req2} ->

src/nrte_publications.erl

@@ -0,0 +1,37 @@
+%%% Copyright 2025 Nomasystems, S.L. http://www.nomasystems.com
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License
+-module(nrte_publications).
+
+%%% EXTERNAL EXPORTS
+-export([subscription_init_link_terminate/2]).
+
+%%% MACROS
+-define(TOPIC_SEPARATOR, <<";">>).
+
+%%%-----------------------------------------------------------------------------
+%%% EXTERNAL EXPORTS
+%%%-----------------------------------------------------------------------------
+-spec subscription_init_link_terminate(binary(), list(binary())) -> ok.
+subscription_init_link_terminate(Source, TopicList) ->
+    Topics = binary:join(TopicList, ?TOPIC_SEPARATOR),
+    nrte:publish(<<"nrte:subscription_init:", Source/binary>>, Topics),
+    Pid = self(),
+    spawn_link(fun() ->
+        process_flag(trap_exit, true),
+        receive
+            {'EXIT', Pid, _} ->
+                nrte:publish(<<"nrte:subscription_terminate:", Source/binary>>, Topics)
+        end
+    end),
+    ok.

src/nrte_websocket.erl

@@ -24,7 +24,8 @@ init(Req, Opts) ->
     case nrte_auth:authorization(Req, subscribe) of
         {authorized, TopicList} ->
             nrte:subscribe(TopicList),
-            {cowboy_websocket, Req, []};
+            nrte_publications:subscription_init_link_terminate(<<"websocket">>, TopicList),
+            {cowboy_websocket, Req, Opts};
         {unauthorized, Req2} ->
             {stop, Req2, Opts}
     end.

test/nrte_eventsource_SUITE.erl

@@ -26,6 +26,8 @@ all() ->
         connect_authorized_pattern,
         connect_forbidden_pattern,
         connect_unauthorized,
+        receive_subscription_init,
+        receive_subscription_terminate,
         receive_topic_message
     ].
 
@@ -81,6 +83,36 @@ connect_unauthorized(_Conf) ->
     application:unset_env(nrte, auth_type),
     ok.
 
+receive_subscription_init() ->
+    [{userdata, [{doc, "Tests receiving the special subscription init message"}]}].
+
+receive_subscription_init(_Conf) ->
+    Subscription = <<"nrte:subscription_init:eventsource">>,
+    ok = nrte:subscribe([Subscription]),
+    Topics = <<"topic1;topic2;topic3">>,
+    {ok, {Pid, _StreamRef}} = connect_es(Topics),
+    ExpectedMessage = <<Subscription/binary, ";", Topics/binary>>,
+    receive
+        {nrte_message, ExpectedMessage} -> ok
+    after 1000 -> throw(timeout)
+    end,
+    gun:close(Pid).
+
+receive_subscription_terminate() ->
+    [{userdata, [{doc, "Tests receiving the special subscription terminate message"}]}].
+
+receive_subscription_terminate(_Conf) ->
+    Subscription = <<"nrte:subscription_terminate:eventsource">>,
+    ok = nrte:subscribe([Subscription]),
+    Topics = <<"topic1;topic2;topic3">>,
+    {ok, {Pid, _StreamRef}} = connect_es(Topics),
+    gun:close(Pid),
+    ExpectedMessage = <<Subscription/binary, ";", Topics/binary>>,
+    receive
+        {nrte_message, ExpectedMessage} -> ok
+    after 1000 -> throw(timeout)
+    end.
+
 receive_topic_message() ->
     [{userdata, [{doc, "Tests receiving a published topic message"}]}].
 

test/nrte_websocket_SUITE.erl

@@ -30,6 +30,8 @@ all() ->
         connect_forbidden_subscription,
         connect_unauthorized,
         coverage_completion,
+        receive_subscription_init,
+        receive_subscription_terminate,
         receive_topic_message,
         send_text,
         send_unsupported_text
@@ -93,6 +95,34 @@ coverage_completion() ->
 coverage_completion(_Conf) ->
     _ = nrte_websocket:websocket_info(dummy, state).
 
+receive_subscription_init() ->
+    [{userdata, [{doc, "Tests receiving the special subscription init message"}]}].
+
+receive_subscription_init(_Conf) ->
+    Subscription = <<"nrte:subscription_init:websocket">>,
+    ok = nrte:subscribe([Subscription]),
+    {ok, {Pid, _StreamRef}} = connect_ws(),
+    ExpectedMessage = <<Subscription/binary, ";", ?TOPIC>>,
+    receive
+        {nrte_message, ExpectedMessage} -> ok
+    after 1000 -> throw(timeout)
+    end,
+    gun:close(Pid).
+
+receive_subscription_terminate() ->
+    [{userdata, [{doc, "Tests receiving the special subscription terminate message"}]}].
+
+receive_subscription_terminate(_Conf) ->
+    Subscription = <<"nrte:subscription_terminate:websocket">>,
+    ok = nrte:subscribe([Subscription]),
+    {ok, {Pid, _StreamRef}} = connect_ws(),
+    gun:close(Pid),
+    ExpectedMessage = <<Subscription/binary, ";", ?TOPIC>>,
+    receive
+        {nrte_message, ExpectedMessage} -> ok
+    after 1000 -> throw(timeout)
+    end.
+
 receive_topic_message() ->
     [{userdata, [{doc, "Tests receiving a published topic message"}]}].