From 852b159dfc4376b91459617ed34fcdbdc181c099 Mon Sep 17 00:00:00 2001 From: Marcelino Alberdi Pereira Date: Tue, 30 Dec 2025 12:03:27 +0100 Subject: [PATCH] feat(#15): Allow receiving callbacks when a client connects/disconnects Closes #15. --- README.md | 11 ++++++++++ src/nrte_eventsource.erl | 1 + src/nrte_publications.erl | 37 +++++++++++++++++++++++++++++++++ src/nrte_websocket.erl | 3 ++- test/nrte_eventsource_SUITE.erl | 32 ++++++++++++++++++++++++++++ test/nrte_websocket_SUITE.erl | 30 ++++++++++++++++++++++++++ 6 files changed, 113 insertions(+), 1 deletion(-) create mode 100644 src/nrte_publications.erl diff --git a/README.md b/README.md index d462bbd..4469327 100644 --- a/README.md +++ b/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). diff --git a/src/nrte_eventsource.erl b/src/nrte_eventsource.erl index 4cf6538..6e135e9 100644 --- a/src/nrte_eventsource.erl +++ b/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} -> diff --git a/src/nrte_publications.erl b/src/nrte_publications.erl new file mode 100644 index 0000000..672f603 --- /dev/null +++ b/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. diff --git a/src/nrte_websocket.erl b/src/nrte_websocket.erl index a366022..0fec675 100644 --- a/src/nrte_websocket.erl +++ b/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. diff --git a/test/nrte_eventsource_SUITE.erl b/test/nrte_eventsource_SUITE.erl index 709a6f8..71789ed 100644 --- a/test/nrte_eventsource_SUITE.erl +++ b/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 = <>, + 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 = <>, + receive + {nrte_message, ExpectedMessage} -> ok + after 1000 -> throw(timeout) + end. + receive_topic_message() -> [{userdata, [{doc, "Tests receiving a published topic message"}]}]. diff --git a/test/nrte_websocket_SUITE.erl b/test/nrte_websocket_SUITE.erl index 9693677..069d080 100644 --- a/test/nrte_websocket_SUITE.erl +++ b/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 = <>, + 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 = <>, + receive + {nrte_message, ExpectedMessage} -> ok + after 1000 -> throw(timeout) + end. + receive_topic_message() -> [{userdata, [{doc, "Tests receiving a published topic message"}]}].