<commit_before>%%
%%    Wave - MQTT Broker
%%    Copyright (C) 2014-2016 - <NAME>
%%
%%    This program is free software: you can redistribute it and/or modify
%%    it under the terms of the GNU Affero General Public License as published
%%    by the Free Software Foundation, version 3 of the License.
%%
%%    This program is distributed in the hope that it will be useful,
%%    but WITHOUT ANY WARRANTY; without even the implied warranty of
%%    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
%%    GNU Affero General Public License for more details.
%%
%%    You should have received a copy of the GNU Affero General Public License
%%    along with this program.  If not, see <http://www.gnu.org/licenses/>.

-module(mqtt_retain).
-author("<NAME> <<EMAIL>>").
-behaviour(gen_fsm).

%%
%% NOTE: this module is mixing plain funs and fsm behaviour
%%       should it be splitted ?
%% NOTE: gen_fsm is doing same thing as mqtt_lastwill_session (fake publisher session)
%%       should we merge both ?
%%       => mqtt_loopback_session

-include("mqtt_msg.hrl").

% public API
-export([start_link/0, store/1, publish/3]).
% gen_fsm funs
-export([init/1, handle_event/3, handle_sync_event/4, handle_info/3, terminate/3, code_change/4]).
% gen_fsm internals
-export([run/2]).

-spec start_link() -> {ok, pid()} | ignore | {error, any()}.
start_link() ->
    gen_fsm:start_link({local, retain_session}, ?MODULE, [], []).

init([]) ->
    {ok, run, undefined}.

%%
%% PUBLIC API
%%

-spec store(mqtt_msg()) -> pass|retained.
store(#mqtt_msg{type='PUBLISH', retain=0}) ->
    % non-retainable message
    pass;
store(#mqtt_msg{type='PUBLISH', retain=1, qos=Qos, payload=Payload}) ->
    Topic = proplists:get_value(topic, Payload),
    Data  = proplists:get_value(data, Payload),

    store2(Topic, Data, Qos),
    retained.

% empty payload: delete retained message
store2(Topic, <<>>, _) ->
    lager:debug("deleting retained message under ~p", [Topic]),
    wave_db:del(<<"retain:", Topic/binary>>),
    ok;
store2(Topic, Data, Qos) ->
    lager:debug("saving retained message (t= ~p, m=~p)", [Topic, Data]),
    wave_db:set({h, <<"retain:", Topic/binary>>}, #{data => Data, qos => Qos}),
    ok.


-spec publish(mqtt_topic_registry:subscriber(), mqtt_topic(), mqtt_qos()) -> ok.
publish(Subscriber, TopicF, Qos) ->
    % "/foo/+/bar/#" => "/foo/*/bar/*"
    RTopic     = topic2redis(TopicF, <<"retain:">>),
    {ok, Keys} = wave_db:search(RTopic),
    Keys2      = lists:filter(fun(<<"retain:", T/binary>>) -> topicmatch(TopicF, T) end, Keys),

    lager:debug("found retained topics: ~p", Keys2),
    publish2(Keys2, {TopicF, Qos, Subscriber, []}).

publish2(                                        [], _) ->
    ok;
publish2([Match = <<"retain:", Topic/binary>>|Keys], Subscription) ->
    {ok, Values} = wave_db:get({h, Match}),

    Data = proplists:get_value(<<"data">>, Values),
    Qos  = wave_utils:int(proplists:get_value(<<"qos">> , Values, "0")),
    Msg  = #mqtt_msg{type='PUBLISH', qos=Qos, retain=1, payload=[{topic, Topic}, {data, Data}]},

    {ok, MsgWorker} = mqtt_message_worker:start_link(),
    mqtt_message_worker:publish(MsgWorker, retain_session, Msg, [Subscription]), 

    publish2(Keys, Subscription).

% NOTE: specific to redis, needs abstraction<commit_msg>[fix] parameter in brackets in lager:debug()<commit_after>%%
%%    Wave - MQTT Broker
%%    Copyright (C) 2014-2016 - <NAME>
%%
%%    This program is free software: you can redistribute it and/or modify
%%    it under the terms of the GNU Affero General Public License as published
%%    by the Free Software Foundation, version 3 of the License.
%%
%%    This program is distributed in the hope that it will be useful,
%%    but WITHOUT ANY WARRANTY; without even the implied warranty of
%%    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
%%    GNU Affero General Public License for more details.
%%
%%    You should have received a copy of the GNU Affero General Public License
%%    along with this program.  If not, see <http://www.gnu.org/licenses/>.

-module(mqtt_retain).
-author("<NAME> <<EMAIL>>").
-behaviour(gen_fsm).

%%
%% NOTE: this module is mixing plain funs and fsm behaviour
%%       should it be splitted ?
%% NOTE: gen_fsm is doing same thing as mqtt_lastwill_session (fake publisher session)
%%       should we merge both ?
%%       => mqtt_loopback_session

-include("mqtt_msg.hrl").

% public API
-export([start_link/0, store/1, publish/3]).
% gen_fsm funs
-export([init/1, handle_event/3, handle_sync_event/4, handle_info/3, terminate/3, code_change/4]).
% gen_fsm internals
-export([run/2]).

-spec start_link() -> {ok, pid()} | ignore | {error, any()}.
start_link() ->
    gen_fsm:start_link({local, retain_session}, ?MODULE, [], []).

init([]) ->
    {ok, run, undefined}.

%%
%% PUBLIC API
%%

-spec store(mqtt_msg()) -> pass|retained.
store(#mqtt_msg{type='PUBLISH', retain=0}) ->
    % non-retainable message
    pass;
store(#mqtt_msg{type='PUBLISH', retain=1, qos=Qos, payload=Payload}) ->
    Topic = proplists:get_value(topic, Payload),
    Data  = proplists:get_value(data, Payload),

    store2(Topic, Data, Qos),
    retained.

% empty payload: delete retained message
store2(Topic, <<>>, _) ->
    lager:debug("deleting retained message under ~p", [Topic]),
    wave_db:del(<<"retain:", Topic/binary>>),
    ok;
store2(Topic, Data, Qos) ->
    lager:debug("saving retained message (t= ~p, m=~p)", [Topic, Data]),
    wave_db:set({h, <<"retain:", Topic/binary>>}, #{data => Data, qos => Qos}),
    ok.


-spec publish(mqtt_topic_registry:subscriber(), mqtt_topic(), mqtt_qos()) -> ok.
publish(Subscriber, TopicF, Qos) ->
    % "/foo/+/bar/#" => "/foo/*/bar/*"
    RTopic     = topic2redis(TopicF, <<"retain:">>),
    {ok, Keys} = wave_db:search(RTopic),
    Keys2      = lists:filter(fun(<<"retain:", T/binary>>) -> topicmatch(TopicF, T) end, Keys),

    lager:debug("found retained topics: ~p", [Keys2]),
    publish2(Keys2, {TopicF, Qos, Subscriber, []}).

publish2(                                        [], _) ->
    ok;
publish2([Match = <<"retain:", Topic/binary>>|Keys], Subscription) ->
    {ok, Values} = wave_db:get({h, Match}),

    Data = proplists:get_value(<<"data">>, Values),
    Qos  = wave_utils:int(proplists:get_value(<<"qos">> , Values, "0")),
    Msg  = #mqtt_msg{type='PUBLISH', qos=Qos, retain=1, payload=[{topic, Topic}, {data, Data}]},

    {ok, MsgWorker} = mqtt_message_worker:start_link(),
    mqtt_message_worker:publish(MsgWorker, retain_session, Msg, [Subscription]), 

    publish2(Keys, Subscription).

% NOTE: specific to redis, needs abstraction
{'fraction_non_alphanumeric': 0.1692029531414796, 'fraction_numerical': 0.010245592888353172, 'mean_word_length': 3.845255474452555, 'pattern_counts': {'":': 0, '<': 45, '<?xml version=': 0, '>': 79, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4095423', 'n_tokens_mistral': 2336, 'n_tokens_neox': 2255, 'n_words': 600}