<commit_before>%%%-------------------------------------------------------------------
%%% @author <NAME>
%%% @copyright (C) 2015 ACK CYFRONET AGH
%%% This software is released under the MIT license
%%% cited in 'LICENSE.txt'.
%%% @end
%%%-------------------------------------------------------------------
%%% @doc
%%% This module implements gen_fsm behaviour and is responsible for sorting
%%% incoming stream messages in ascending order of sequence number and
%%% forwarding them to the router. It is supervised by sequencer stream
%%% supervisor and coordinated by sequencer manager.
%%% @end
%%%-------------------------------------------------------------------
-module(sequencer_in_stream).
-author("<NAME>").

-behaviour(gen_fsm).

-include("global_definitions.hrl").
-include("proto/oneclient/client_messages.hrl").
-include_lib("ctool/include/logging.hrl").

%% API
-export([start_link/3]).

%% gen_fsm callbacks
-export([init/1, handle_event/3, handle_sync_event/4, handle_info/3, terminate/3,
    code_change/4]).

%% gen_fsm states
-export([receiving/2, processing/2, requesting/2]).

-type stream_id() :: sequencer:stream_id().
-type sequence_number() :: sequencer:sequence_number().

%% sequencer in stream state:
%% session_id           - ID of a session associated with this sequencer stream
%% sequencer_manager    - pid of a sequencer manager that controls this
%%                        sequencer stream
%% stream_id            - ID of a communication stream associated with this
%%                        sequencer stream
%% sequence_number      - sequence number of message that can be forwarded
%% sequence_number_ack  - sequence number of last acknowledged message
%% messages             - mapping from sequence number to message for messages
%%                        waiting to be forwarded
-record(state, {
    session_id :: session:id(),
    proxy_session_id :: undefined | session:id(),
    sequencer_manager :: pid(),
    stream_id :: stream_id(),
    sequence_number = 0 :: sequence_number(),
    sequence_number_ack = -1 :: -1 | sequence_number(),
    messages = #{} :: #{sequence_number() => #client_message{}},
    is_proxy :: boolean()
}).

-define(MSG_ACK_THRESHOLD, application:get_env(?APP_NAME,
    sequencer_stream_msg_ack_threshold, 100)).
-define(RECEIVING_TIMEOUT, timer:seconds(application:get_env(?APP_NAME,
    sequencer_stream_msg_req_long_timeout_seconds, 10))).
-define(REQUESTING_TIMEOUT, timer:seconds(application:get_env(?APP_NAME,
    sequencer_stream_msg_req_short_timeout_seconds, 1))).

%%%===================================================================
%%% API
%%%===================================================================

%%--------------------------------------------------------------------
%% @doc
%% Starts the sequencer stream for incoming messages.
%% @end
%%--------------------------------------------------------------------
-spec start_link(SeqMan :: pid(), StmId :: stream_id(), SessId :: session:id()) ->
    {ok, SeqStm :: pid()} | ignore | {error, Reason :: term()}.
start_link(SeqMan, StmId, SessId) ->
    gen_fsm:start_link(?MODULE, [SeqMan, StmId, SessId], []).

%%%===================================================================
%%% gen_fsm callbacks
%%%===================================================================

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Initializes the sequencer stream for incoming messages.
%% @end
%%--------------------------------------------------------------------
-spec(init(Args :: term()) ->
    {ok, StateName :: atom(), StateData :: #state{}} |
    {ok, StateName :: atom(), StateData :: #state{}, timeout() | hibernate} |
    {stop, Reason :: term()} | ignore).
init([SeqMan, StmId, SessId]) ->
    ?debug("Initializing sequencer in stream for session ~p", [SessId]),
    process_flag(trap_exit, true),
    register_stream(SeqMan, StmId),
    {ok, #document{value = #session{type = SessionType, proxy_via = ProxyVia}}} = session:get(SessId),
    IsProxy = SessionType =:= provider_incoming orelse SessionType =:= provider_outgoing orelse ProxyVia =/= undefined,
    send_message_stream_reset(StmId, SessId, IsProxy),
    {ok, receiving, #state{
        sequencer_manager = SeqMan,
        session_id = SessId,
        stream_id = StmId,
        is_proxy = IsProxy
    }, ?RECEIVING_TIMEOUT}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Whenever a gen_fsm receives an event sent using
%% gen_fsm:send_all_state_event/2, this function is called to handle
%% the event.
%% @end
%%--------------------------------------------------------------------
-spec(handle_event(Event :: term(), StateName :: atom(),
    StateData :: #state{}) ->
    {next_state, NextStateName :: atom(), NewStateData :: #state{}} |
    {next_state, NextStateName :: atom(), NewStateData :: #state{},
        timeout() | hibernate} |
    {stop, Reason :: term(), NewStateData :: #state{}}).
handle_event(Event, StateName, State) ->
    ?log_bad_request({Event, StateName}),
    {next_state, StateName, State}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Whenever a gen_fsm receives an event sent using
%% gen_fsm:sync_send_all_state_event/[2,3], this function is called
%% to handle the event.
%% @end
%%--------------------------------------------------------------------
-spec(handle_sync_event(Event :: term(), From :: {pid(), Tag :: term()},
    StateName :: atom(), StateData :: term()) ->
    {reply, Reply :: term(), NextStateName :: atom(), NewStateData :: term()} |
    {reply, Reply :: term(), NextStateName :: atom(), NewStateData :: term(),
        timeout() | hibernate} |
    {next_state, NextStateName :: atom(), NewStateData :: term()} |
    {next_state, NextStateName :: atom(), NewStateData :: term(),
        timeout() | hibernate} |
    {stop, Reason :: term(), Reply :: term(), NewStateData :: term()} |
    {stop, Reason :: term(), NewStateData :: term()}).
handle_sync_event(Event, From, StateName, State) ->
    ?log_bad_request({Event, From, StateName}),
    {next_state, StateName, State}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% This function is called by a gen_fsm when it receives any
%% message other than a synchronous or asynchronous event
%% (or a system message).
%% @end
%%--------------------------------------------------------------------
-spec(handle_info(Info :: term(), StateName :: atom(),
    StateData :: term()) ->
    {next_state, NextStateName :: atom(), NewStateData :: term()} |
    {next_state, NextStateName :: atom(), NewStateData :: term(),
        timeout() | hibernate} |
    {stop, Reason :: normal | term(), NewStateData :: term()}).
handle_info({'EXIT', _, shutdown}, _, State) ->
    {stop, normal, State};

handle_info(Info, StateName, State) ->
    ?log_bad_request({Info, StateName}),
    {next_state, StateName, State}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% This function is called by a gen_fsm when it is about to
%% terminate. It should be the opposite of Module:init/1 and do any
%% necessary cleaning up. When it returns, the gen_fsm terminates with
%% Reason. The return value is ignored.
%%
%% @end
%%--------------------------------------------------------------------
-spec(terminate(Reason :: normal | shutdown | {shutdown, term()}
| term(), StateName :: atom(), StateData :: term()) -> term()).
terminate(Reason, StateName, #state{stream_id = StmId, sequence_number = SeqNum,
    session_id = SessId, sequencer_manager = SeqMan, is_proxy = IsProxy} = State) ->
    ?log_terminate(Reason, {StateName, State}),
    Msg = #message_acknowledgement{stream_id = StmId, sequence_number = SeqNum - 1},
    CommunicatorModule = communicator_module(IsProxy),
    case CommunicatorModule:send(Msg, SessId) of
        ok -> ok;
        {error, Reason} -> SeqMan ! {send, Msg, SessId}
    end,
    unregister_stream(State).<commit_msg>VFS-4158 Fix batmatch in sequencer<commit_after>%%%-------------------------------------------------------------------
%%% @author <NAME>
%%% @copyright (C) 2015 ACK CYFRONET AGH
%%% This software is released under the MIT license
%%% cited in 'LICENSE.txt'.
%%% @end
%%%-------------------------------------------------------------------
%%% @doc
%%% This module implements gen_fsm behaviour and is responsible for sorting
%%% incoming stream messages in ascending order of sequence number and
%%% forwarding them to the router. It is supervised by sequencer stream
%%% supervisor and coordinated by sequencer manager.
%%% @end
%%%-------------------------------------------------------------------
-module(sequencer_in_stream).
-author("<NAME>").

-behaviour(gen_fsm).

-include("global_definitions.hrl").
-include("proto/oneclient/client_messages.hrl").
-include_lib("ctool/include/logging.hrl").

%% API
-export([start_link/3]).

%% gen_fsm callbacks
-export([init/1, handle_event/3, handle_sync_event/4, handle_info/3, terminate/3,
    code_change/4]).

%% gen_fsm states
-export([receiving/2, processing/2, requesting/2]).

-type stream_id() :: sequencer:stream_id().
-type sequence_number() :: sequencer:sequence_number().

%% sequencer in stream state:
%% session_id           - ID of a session associated with this sequencer stream
%% sequencer_manager    - pid of a sequencer manager that controls this
%%                        sequencer stream
%% stream_id            - ID of a communication stream associated with this
%%                        sequencer stream
%% sequence_number      - sequence number of message that can be forwarded
%% sequence_number_ack  - sequence number of last acknowledged message
%% messages             - mapping from sequence number to message for messages
%%                        waiting to be forwarded
-record(state, {
    session_id :: session:id(),
    proxy_session_id :: undefined | session:id(),
    sequencer_manager :: pid(),
    stream_id :: stream_id(),
    sequence_number = 0 :: sequence_number(),
    sequence_number_ack = -1 :: -1 | sequence_number(),
    messages = #{} :: #{sequence_number() => #client_message{}},
    is_proxy :: boolean()
}).

-define(MSG_ACK_THRESHOLD, application:get_env(?APP_NAME,
    sequencer_stream_msg_ack_threshold, 100)).
-define(RECEIVING_TIMEOUT, timer:seconds(application:get_env(?APP_NAME,
    sequencer_stream_msg_req_long_timeout_seconds, 10))).
-define(REQUESTING_TIMEOUT, timer:seconds(application:get_env(?APP_NAME,
    sequencer_stream_msg_req_short_timeout_seconds, 1))).

%%%===================================================================
%%% API
%%%===================================================================

%%--------------------------------------------------------------------
%% @doc
%% Starts the sequencer stream for incoming messages.
%% @end
%%--------------------------------------------------------------------
-spec start_link(SeqMan :: pid(), StmId :: stream_id(), SessId :: session:id()) ->
    {ok, SeqStm :: pid()} | ignore | {error, Reason :: term()}.
start_link(SeqMan, StmId, SessId) ->
    gen_fsm:start_link(?MODULE, [SeqMan, StmId, SessId], []).

%%%===================================================================
%%% gen_fsm callbacks
%%%===================================================================

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Initializes the sequencer stream for incoming messages.
%% @end
%%--------------------------------------------------------------------
-spec(init(Args :: term()) ->
    {ok, StateName :: atom(), StateData :: #state{}} |
    {ok, StateName :: atom(), StateData :: #state{}, timeout() | hibernate} |
    {stop, Reason :: term()} | ignore).
init([SeqMan, StmId, SessId]) ->
    ?debug("Initializing sequencer in stream for session ~p", [SessId]),
    process_flag(trap_exit, true),
    register_stream(SeqMan, StmId),
    {ok, #document{value = #session{type = SessionType, proxy_via = ProxyVia}}} = session:get(SessId),
    IsProxy = SessionType =:= provider_incoming orelse SessionType =:= provider_outgoing orelse ProxyVia =/= undefined,
    send_message_stream_reset(StmId, SessId, IsProxy),
    {ok, receiving, #state{
        sequencer_manager = SeqMan,
        session_id = SessId,
        stream_id = StmId,
        is_proxy = IsProxy
    }, ?RECEIVING_TIMEOUT}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Whenever a gen_fsm receives an event sent using
%% gen_fsm:send_all_state_event/2, this function is called to handle
%% the event.
%% @end
%%--------------------------------------------------------------------
-spec(handle_event(Event :: term(), StateName :: atom(),
    StateData :: #state{}) ->
    {next_state, NextStateName :: atom(), NewStateData :: #state{}} |
    {next_state, NextStateName :: atom(), NewStateData :: #state{},
        timeout() | hibernate} |
    {stop, Reason :: term(), NewStateData :: #state{}}).
handle_event(Event, StateName, State) ->
    ?log_bad_request({Event, StateName}),
    {next_state, StateName, State}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Whenever a gen_fsm receives an event sent using
%% gen_fsm:sync_send_all_state_event/[2,3], this function is called
%% to handle the event.
%% @end
%%--------------------------------------------------------------------
-spec(handle_sync_event(Event :: term(), From :: {pid(), Tag :: term()},
    StateName :: atom(), StateData :: term()) ->
    {reply, Reply :: term(), NextStateName :: atom(), NewStateData :: term()} |
    {reply, Reply :: term(), NextStateName :: atom(), NewStateData :: term(),
        timeout() | hibernate} |
    {next_state, NextStateName :: atom(), NewStateData :: term()} |
    {next_state, NextStateName :: atom(), NewStateData :: term(),
        timeout() | hibernate} |
    {stop, Reason :: term(), Reply :: term(), NewStateData :: term()} |
    {stop, Reason :: term(), NewStateData :: term()}).
handle_sync_event(Event, From, StateName, State) ->
    ?log_bad_request({Event, From, StateName}),
    {next_state, StateName, State}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% This function is called by a gen_fsm when it receives any
%% message other than a synchronous or asynchronous event
%% (or a system message).
%% @end
%%--------------------------------------------------------------------
-spec(handle_info(Info :: term(), StateName :: atom(),
    StateData :: term()) ->
    {next_state, NextStateName :: atom(), NewStateData :: term()} |
    {next_state, NextStateName :: atom(), NewStateData :: term(),
        timeout() | hibernate} |
    {stop, Reason :: normal | term(), NewStateData :: term()}).
handle_info({'EXIT', _, shutdown}, _, State) ->
    {stop, normal, State};

handle_info(Info, StateName, State) ->
    ?log_bad_request({Info, StateName}),
    {next_state, StateName, State}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% This function is called by a gen_fsm when it is about to
%% terminate. It should be the opposite of Module:init/1 and do any
%% necessary cleaning up. When it returns, the gen_fsm terminates with
%% Reason. The return value is ignored.
%%
%% @end
%%--------------------------------------------------------------------
-spec(terminate(Reason :: normal | shutdown | {shutdown, term()}
| term(), StateName :: atom(), StateData :: term()) -> term()).
terminate(Reason, StateName, #state{stream_id = StmId, sequence_number = SeqNum,
    session_id = SessId, sequencer_manager = SeqMan, is_proxy = IsProxy} = State) ->
    ?log_terminate(Reason, {StateName, State}),
    Msg = #message_acknowledgement{stream_id = StmId, sequence_number = SeqNum - 1},
    CommunicatorModule = communicator_module(IsProxy),
    case CommunicatorModule:send(Msg, SessId) of
        ok -> ok;
        {error, _Reason2} -> SeqMan ! {send, Msg, SessId}
    end,
    unregister_stream(State).
{'fraction_non_alphanumeric': 0.2900909822368014, 'fraction_numerical': 0.0037754533638670543, 'mean_word_length': 2.4408006814310053, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2009318', 'n_tokens_mistral': 4420, 'n_tokens_neox': 4286, 'n_words': 1161}