<commit_before>%%%-------------------------------------------------------------------
%%% @author Kalin
%%% @copyright (C) 2015, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 21. Feb 2015 5:37 PM
%%%-------------------------------------------------------------------
-module(mqtt_session).
-author("Kalin").

%%-include("mqtt_packets.hrl").
-include("mqtt_internal_msgs.hrl").
%% API
-export([append_msg/3,
         append_message_comp/2,
         message_ack/2,
         message_pub_rec/2,
         message_pub_comp/2,
         %%recover/1,
         subscribe/2,
         unsubscribe/2,
         cleanup/1,
         new/2, to_publish/4, to_pubrel/1]).


-record(session_out,{
    client_id                 ::binary(),            %% The id of the client
    is_persistent             ::boolean(),           %% whether the session needs to be persisted
    packet_seq                ::non_neg_integer(),   %% The latest packet id (incremented by 1 for every packet)
    qos1 = dict:new()         ,
    qos2 = dict:new()         ,
    qos2_rec = gb_sets:new()  ,
    refs = gb_sets:new()
%%     subs = orddict:new()      ,
%%     min_subs = orddict:new()
}).


%% ================================================================================
%% SUBSCRIPTIONS
%% ================================================================================

subscribe(S =#session_out{ client_id = ClientId},NewSubs) ->

    %% Maintain a flat list of subscriptions
%%     Subs1 = lists:foldr(fun({Topic,QoS},Acc) ->
%%                                 orddict:store(Topic,QoS,Acc)
%%                               end,
%%                              Subs, NewSubs),
%%     MinCover = mqtt_topic:min_cover(orddict:to_list(Subs1)),
%%     _MinCover =
    [
        begin
            error_logger:info_msg("Subscribing: ~p,~p,~p,~n",[ClientId,Topic,QoS]),
            mqtt_sub_repo:add_sub(ClientId,Topic,QoS)
        end
        || {Topic,QoS} <- NewSubs
    ]
    
    %% S#session_out{subs = Subs1, min_subs = MinCover}
    %% @todo: Deduplicate, persist
.

unsubscribe(S = #session_out{subs = Subs, client_id = ClientId},OldSubs) ->
    [
        mqtt_sub_repo:remove_sub(ClientId,Topic)
        || Topic <- Subs
    ]
%%     S#session_out{subs =
%%                   lists:foldr(fun(Topic,Acc) ->
%%                                 orddict:erase(Topic,Acc)
%%                               end,
%%                               Subs, OldSubs)}
    %% @todo: Deduplicate, persist
    .

%% get_sub_diff(Subs,NewSubs) ->
%%     orddict:update().

%% check_sub_exists(Subs,NewSub = {Filter,QoS}) ->
%%     case lists:dropwhile(fun(Sub) -> not mqtt_topic:is_covered_by(Sub, NewSub) end, Subs) of
%%         [] -> error;
%%         [Sub| _] = Sub
%%     end


%% e.g. at shutdown with ClearSession = false
cleanup(#session_out{subs = Subs, client_id = ClientId, is_persistent = false}) ->
    mqtt_reg_repo:unregister(self(),ClientId),
    [ mqtt_sub_repo:remove_sub(ClientId,Topic) || {Topic,_QoS}  <- Subs ];

cleanup(_S) ->
    ok.


%%add_or_replace_sub(ClientId,{Topic,QoS},Subs)->
%%     State = case orddict:find(Topic,Subs) of
%%                 error ->
%%                     new;
%%                 {ok,CurrentQoS} when CurrentQoS =/= QoS ->
%%                     replaced;
%%                 {ok,CurrentQoS} when CurrentQoS =:= QoS ->
%%                     exists
%%             end,
%%     mqtt_sub_repo:add_sub(ClientId,Topic,QoS),
%%     {State,orddict:store(Topic,QoS,Subs)}.
<commit_msg>Restoring usage of subscription collection<commit_after>%%%-------------------------------------------------------------------
%%% @author Kalin
%%% @copyright (C) 2015, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 21. Feb 2015 5:37 PM
%%%-------------------------------------------------------------------
-module(mqtt_session).
-author("Kalin").

%%-include("mqtt_packets.hrl").
-include("mqtt_internal_msgs.hrl").
%% API
-export([append_msg/3,
         append_message_comp/2,
         message_ack/2,
         message_pub_rec/2,
         message_pub_comp/2,
         %%recover/1,
         subscribe/2,
         unsubscribe/2,
         cleanup/1,
         new/2, to_publish/4, to_pubrel/1]).


-record(session_out,{
    client_id                 ::binary(),            %% The id of the client
    is_persistent             ::boolean(),           %% whether the session needs to be persisted
    packet_seq                ::non_neg_integer(),   %% The latest packet id (incremented by 1 for every packet)
    qos1 = dict:new()         ,
    qos2 = dict:new()         ,
    qos2_rec = gb_sets:new()  ,
    refs = gb_sets:new()      ,
    subs = orddict:new()
%%     min_subs = orddict:new()
}).


%% ================================================================================
%% SUBSCRIPTIONS
%% ================================================================================

subscribe(S =#session_out{subs = Subs, client_id = ClientId},NewSubs) ->

    %% Maintain a flat list of subscriptions
    %% @todo: Optimize/deduplicate
    Subs1 = lists:foldr(fun({Topic,QoS},Acc) ->
                                orddict:store(Topic,QoS,Acc)
                              end,
                             Subs, NewSubs),
    [
        begin
            error_logger:info_msg("Subscribing: ~p,~p,~p,~n",[ClientId,Topic,QoS]),
            mqtt_sub_repo:add_sub(ClientId,Topic,QoS)
        end
        || {Topic,QoS} <- NewSubs
    ],

    %% @todo: Deduplicate, persist
    S#session_out{subs = Subs1}.

unsubscribe(S = #session_out{subs = Subs, client_id = ClientId},OldSubs) ->
    [
        mqtt_sub_repo:remove_sub(ClientId,Topic)
        || Topic <- OldSubs
    ],
    S#session_out{subs =
                  lists:foldr(fun(Topic,Acc) ->
                                orddict:erase(Topic,Acc)
                              end,
                              Subs, OldSubs)}.

%% get_sub_diff(Subs,NewSubs) ->
%%     orddict:update().

%% check_sub_exists(Subs,NewSub = {Filter,QoS}) ->
%%     case lists:dropwhile(fun(Sub) -> not mqtt_topic:is_covered_by(Sub, NewSub) end, Subs) of
%%         [] -> error;
%%         [Sub| _] = Sub
%%     end


%% clear persisted data during shutdown
cleanup(S = #session_out{client_id = ClientId}) ->
    mqtt_reg_repo:unregister(ClientId),
    maybe_clear_subs(S).

maybe_clear_subs(#session_out{subs = Subs, client_id = ClientId, is_persistent = true}) ->
    [ mqtt_sub_repo:remove_sub(ClientId,Topic) || {Topic,_QoS}  <- Subs ],
    ok;

maybe_clear_subs(_S) ->
    ok.



%%add_or_replace_sub(ClientId,{Topic,QoS},Subs)->
%%     State = case orddict:find(Topic,Subs) of
%%                 error ->
%%                     new;
%%                 {ok,CurrentQoS} when CurrentQoS =/= QoS ->
%%                     replaced;
%%                 {ok,CurrentQoS} when CurrentQoS =:= QoS ->
%%                     exists
%%             end,
%%     mqtt_sub_repo:add_sub(ClientId,Topic,QoS),
%%     {State,orddict:store(Topic,QoS,Subs)}.

{'fraction_non_alphanumeric': 0.2212656048213517, 'fraction_numerical': 0.009040034438226431, 'mean_word_length': 1.9483925549915397, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 34, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5157665', 'n_tokens_mistral': 2365, 'n_tokens_neox': 2198, 'n_words': 359}