<commit_before>%%%-------------------------------------------------------------------
%% @copyright <NAME>
%% @author <NAME> <<EMAIL>>
%% @version {@vsn}, {@date} {@time}
%% @doc Syslog/tcp drain
%% @end
%%%-------------------------------------------------------------------

-module(logplex_tcpsyslog_drain2).
-behaviour(gen_fsm).
-define(SERVER, ?MODULE).
-define(RECONNECT_MSG, reconnect).
-define(TARGET_SEND_SIZE, 4096).

-include("logplex.hrl").
-include("logplex_logging.hrl").
-include_lib("eunit/include/eunit.hrl").

-record(state, {drain_id :: logplex_drain:id(),
                drain_tok :: logplex_drain:token(),
                channel_id :: logplex_channel:id(),
                host :: string() | inet:ip_address() | binary(),
                port :: inet:port_number(),
                sock = undefined :: 'undefined' | inet:socket(),
                %% Buffer for messages while disconnected
                buf = logplex_drain_buffer:new() :: logplex_drain_buffer:buf(),
                %% Last time we connected or successfully sent data
                last_good_time :: 'undefined' | erlang:timestamp(),
                %% TCP failures since last_good_time
                failures = 0 :: non_neg_integer(),
                %% Reconnect timer reference
                reconnect_tref = undefined :: 'undefined' | reference(),
                %% Send timer reference
                send_tref = undefined :: 'undefined' | reference(),
                connect_time :: 'undefined' | erlang:timestamp()
               }).

%% -type pstate() :: 'disconnected' | 'ready_to_send' | 'sending'.

%% ------------------------------------------------------------------
%% API Function Exports
%% ------------------------------------------------------------------

-export([start_link/5]).

%% ------------------------------------------------------------------
%% gen_fsm Function Exports
%% ------------------------------------------------------------------

-export([disconnected/2,
         ready_to_send/2,
         sending/2
         ]).

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

%% ------------------------------------------------------------------
%% API Function Definitions
%% ------------------------------------------------------------------

start_link(ChannelID, DrainID, DrainTok, Host, Port) ->
    gen_fsm:start_link(?MODULE,
                       [#state{drain_id=DrainID,
                               drain_tok=DrainTok,
                               channel_id=ChannelID,
                               host=Host,
                               port=Port}],
                       []).

%% ------------------------------------------------------------------
%% gen_fsm Function Definitions
%% ------------------------------------------------------------------

%% @private
init([State0 = #state{sock = undefined,
                      host=H, port=P}])
  when H =/= undefined, is_integer(P) ->
    try
        register_with_gproc(State0),
        DrainSize = logplex_app:config(tcp_drain_buffer_size),
        State = State0#state{buf = logplex_drain_buffer:new(DrainSize)},
        ?INFO("drain_id=~p channel_id=~p dest=~s at=spawn",
              log_info(State, [])),
        {ok, disconnected,
         State, hibernate}
    catch
        error:badarg -> ignore
    end.

%% @doc Disconnected state. We wait here for the reconnect timer to
%% fire before initiating the reconnect sequence.
disconnected({timeout, TRef, ?RECONNECT_MSG},
             State = #state{reconnect_tref = TRef, sock = undefined}) ->
    case connect(State) of
        {ok, Sock} ->
            ?INFO("drain_id=~p channel_id=~p dest=~s at=connect try=~p sock=~p",
                  log_info(State, [State#state.failures + 1, Sock])),
            NewState = tcp_good(State#state{sock=Sock,
                                            connect_time=os:timestamp()}),
            send(NewState);
        {error, Reason} ->
            NewState = tcp_bad(State),
            ?ERR("drain_id=~p channel_id=~p dest=~s at=connect "
                 "err=gen_tcp data=~p try=~p last_success=~s",
                 log_info(State, [Reason, NewState#state.failures,
                                  time_failed(NewState)])),
            {next_state, disconnected, reconnect(NewState)}
    end;
disconnected({post, Msg}, State) ->
    {next_state, disconnected,
     reconnect(buffer(Msg, State))};
disconnected(Msg, State) ->
    ?WARN("drain_id=~p channel_id=~p dest=~s err=unexpected_info data=~p",
          log_info(State, [Msg])),
    {next_state, disconnected, State}.

%% @doc We have a socket open and messages to send. Collect up an
%% appropriate amount and flush them to the socket.
ready_to_send({post, Msg}, State) ->
    send(buffer(Msg, State));
ready_to_send({inet_reply, Sock, ok}, S = #state{sock = Sock}) ->
    %% Stale inet reply
    send(S);
ready_to_send(Msg, State) ->
    ?WARN("drain_id=~p channel_id=~p dest=~s err=unexpected_info data=~p",
          log_info(State, [Msg])),
    {next_state, disconnected, State}.


%% @doc We sent some data to the socket and are waiting the result of
%% the send operation.
sending({post, Msg}, State) ->
    {next_state, sending, buffer(Msg, State)};
sending({inet_reply, Sock, ok}, S = #state{sock = Sock}) ->
    send(S);
sending({inet_reply, Sock, {error, Reason}}, S = #state{sock = Sock}) ->
    ?ERR("drain_id=~p channel_id=~p dest=~s state=~p "
         "err=gen_tcp data=~p sock=~p duration=~s",
          log_info(S, [sending, Reason, Sock, duration(S)])),
    {next_state, disconnected, reconnect(tcp_bad(S))};
sending(Msg, State) ->
    ?WARN("drain_id=~p channel_id=~p dest=~s err=unexpected_info data=~p",
          log_info(State, [Msg])),
    {next_state, disconnected, State}.


%% @private
%% state_name(Event, _From, State) ->
%%     ?WARN("[state ~p] Unexpected event ~p",
%%           [state_name, Event]),
%%     {next_state, state_name, State}.

%% @private
handle_event(_Event, StateName, State) ->
    {next_state, StateName, State}.

%% @private
handle_sync_event(Event, _From, StateName, State) ->
    ?WARN("[state ~p] Unexpected event ~p",
          [StateName, Event]),
    {next_state, StateName, State}.

%% @private
handle_info({tcp, S, Data}, StateName, State = #state{sock = S}) ->
    ?WARN("drain_id=~p channel_id=~p dest=~s state=~p"
          "err=unexpected_peer_data data=~p",
          log_info(State, [StateName, Data])),
    {next_state, StateName, State};
handle_info({tcp_error, S, Reason}, StateName, State) ->
    ?ERR("drain_id=~p channel_id=~p dest=~s state=~p "
         "err=gen_tcp data=~p sock=~p duration=~s",
          log_info(State, [StateName, Reason, S, duration(State)])),
    {next_state, disconnected, reconnect(tcp_bad(State))};
handle_info({tcp_closed, S}, StateName, State) ->
    ?INFO("drain_id=~p channel_id=~p dest=~s state=~p "
          "err=gen_tcp data=~p sock=~p duration=~s",
          log_info(State, [StateName, closed, S, duration(State)])),
    {next_state, disconnected, reconnect(tcp_bad(State))};
handle_info(Info, StateName, State) ->
    ?MODULE:StateName(Info, State).

%% @private
terminate(_Reason, _StateName, _State) ->
    ok.

%% @private
code_change(_OldVsn, StateName, State, _Extra) ->
    {ok, StateName, State}.

%% ------------------------------------------------------------------
%% Internal Function Definitions
%% ------------------------------------------------------------------

%% @private
connect(#state{sock = undefined, host=Host, port=Port})
    when is_integer(Port), 0 < Port, Port =< 65535 ->
    SendTimeoutS = logplex_app:config(tcp_syslog_send_timeout_secs),
    HostS = case Host of
                B when is_binary(B) -> binary_to_list(B);
                L when is_list(L) -> L;
                T when is_tuple(T) -> T;
                A when is_atom(A) -> A<commit_msg>Handle async inet errors for all states.<commit_after>%%%-------------------------------------------------------------------
%% @copyright <NAME>
%% @author <NAME> <<EMAIL>>
%% @version {@vsn}, {@date} {@time}
%% @doc Syslog/tcp drain
%% @end
%%%-------------------------------------------------------------------

-module(logplex_tcpsyslog_drain2).
-behaviour(gen_fsm).
-define(SERVER, ?MODULE).
-define(RECONNECT_MSG, reconnect).
-define(TARGET_SEND_SIZE, 4096).

-include("logplex.hrl").
-include("logplex_logging.hrl").
-include_lib("eunit/include/eunit.hrl").

-record(state, {drain_id :: logplex_drain:id(),
                drain_tok :: logplex_drain:token(),
                channel_id :: logplex_channel:id(),
                host :: string() | inet:ip_address() | binary(),
                port :: inet:port_number(),
                sock = undefined :: 'undefined' | inet:socket(),
                %% Buffer for messages while disconnected
                buf = logplex_drain_buffer:new() :: logplex_drain_buffer:buf(),
                %% Last time we connected or successfully sent data
                last_good_time :: 'undefined' | erlang:timestamp(),
                %% TCP failures since last_good_time
                failures = 0 :: non_neg_integer(),
                %% Reconnect timer reference
                reconnect_tref = undefined :: 'undefined' | reference(),
                %% Send timer reference
                send_tref = undefined :: 'undefined' | reference(),
                connect_time :: 'undefined' | erlang:timestamp()
               }).

%% -type pstate() :: 'disconnected' | 'ready_to_send' | 'sending'.

%% ------------------------------------------------------------------
%% API Function Exports
%% ------------------------------------------------------------------

-export([start_link/5]).

%% ------------------------------------------------------------------
%% gen_fsm Function Exports
%% ------------------------------------------------------------------

-export([disconnected/2,
         ready_to_send/2,
         sending/2
         ]).

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

%% ------------------------------------------------------------------
%% API Function Definitions
%% ------------------------------------------------------------------

start_link(ChannelID, DrainID, DrainTok, Host, Port) ->
    gen_fsm:start_link(?MODULE,
                       [#state{drain_id=DrainID,
                               drain_tok=DrainTok,
                               channel_id=ChannelID,
                               host=Host,
                               port=Port}],
                       []).

%% ------------------------------------------------------------------
%% gen_fsm Function Definitions
%% ------------------------------------------------------------------

%% @private
init([State0 = #state{sock = undefined,
                      host=H, port=P}])
  when H =/= undefined, is_integer(P) ->
    try
        register_with_gproc(State0),
        DrainSize = logplex_app:config(tcp_drain_buffer_size),
        State = State0#state{buf = logplex_drain_buffer:new(DrainSize)},
        ?INFO("drain_id=~p channel_id=~p dest=~s at=spawn",
              log_info(State, [])),
        {ok, disconnected,
         State, hibernate}
    catch
        error:badarg -> ignore
    end.

%% @doc Disconnected state. We wait here for the reconnect timer to
%% fire before initiating the reconnect sequence.
disconnected({timeout, TRef, ?RECONNECT_MSG},
             State = #state{reconnect_tref = TRef, sock = undefined}) ->
    case connect(State) of
        {ok, Sock} ->
            ?INFO("drain_id=~p channel_id=~p dest=~s at=connect try=~p sock=~p",
                  log_info(State, [State#state.failures + 1, Sock])),
            NewState = tcp_good(State#state{sock=Sock,
                                            connect_time=os:timestamp()}),
            send(NewState);
        {error, Reason} ->
            NewState = tcp_bad(State),
            ?ERR("drain_id=~p channel_id=~p dest=~s at=connect "
                 "err=gen_tcp data=~p try=~p last_success=~s",
                 log_info(State, [Reason, NewState#state.failures,
                                  time_failed(NewState)])),
            {next_state, disconnected, reconnect(NewState)}
    end;
disconnected({post, Msg}, State) ->
    {next_state, disconnected,
     reconnect(buffer(Msg, State))};
disconnected(Msg, State) ->
    ?WARN("drain_id=~p channel_id=~p dest=~s err=unexpected_info data=~p",
          log_info(State, [Msg])),
    {next_state, disconnected, State}.

%% @doc We have a socket open and messages to send. Collect up an
%% appropriate amount and flush them to the socket.
ready_to_send({post, Msg}, State) ->
    send(buffer(Msg, State));
ready_to_send({inet_reply, Sock, ok}, S = #state{sock = Sock}) ->
    %% Stale inet reply
    send(S);
ready_to_send(Msg, State) ->
    ?WARN("drain_id=~p channel_id=~p dest=~s err=unexpected_info data=~p",
          log_info(State, [Msg])),
    {next_state, disconnected, State}.


%% @doc We sent some data to the socket and are waiting the result of
%% the send operation.
sending({post, Msg}, State) ->
    {next_state, sending, buffer(Msg, State)};
sending({inet_reply, Sock, ok}, S = #state{sock = Sock}) ->
    send(S);
sending({inet_reply, Sock, {error, Reason}}, S = #state{sock = Sock}) ->
    ?ERR("drain_id=~p channel_id=~p dest=~s state=~p "
         "err=gen_tcp data=~p sock=~p duration=~s",
          log_info(S, [sending, Reason, Sock, duration(S)])),
    {next_state, disconnected, reconnect(tcp_bad(S))};
sending(Msg, State) ->
    ?WARN("drain_id=~p channel_id=~p dest=~s err=unexpected_info data=~p",
          log_info(State, [Msg])),
    {next_state, disconnected, State}.


%% @private
%% state_name(Event, _From, State) ->
%%     ?WARN("[state ~p] Unexpected event ~p",
%%           [state_name, Event]),
%%     {next_state, state_name, State}.

%% @private
handle_event(_Event, StateName, State) ->
    {next_state, StateName, State}.

%% @private
handle_sync_event(Event, _From, StateName, State) ->
    ?WARN("[state ~p] Unexpected event ~p",
          [StateName, Event]),
    {next_state, StateName, State}.

%% @private
handle_info({tcp, Sock, Data}, StateName, State = #state{sock = Sock}) ->
    ?WARN("drain_id=~p channel_id=~p dest=~s state=~p"
          "err=unexpected_peer_data data=~p",
          log_info(State, [StateName, Data])),
    {next_state, StateName, State};
handle_info({tcp_error, Sock, Reason}, StateName, State) ->
    ?ERR("drain_id=~p channel_id=~p dest=~s state=~p "
         "err=gen_tcp data=~p sock=~p duration=~s",
          log_info(State, [StateName, Reason, Sock, duration(State)])),
    {next_state, disconnected, reconnect(tcp_bad(State))};
handle_info({inet_reply, Sock, {error, Reason}}, StateName,
            State = #state{sock = Sock}) ->
    ?ERR("drain_id=~p channel_id=~p dest=~s state=~p "
         "err=gen_tcp data=~p sock=~p duration=~s",
          log_info(State, [StateName, Reason, Sock, duration(State)])),
    {next_state, disconnected, reconnect(tcp_bad(State))};
handle_info({tcp_closed, Sock}, StateName, State) ->
    ?INFO("drain_id=~p channel_id=~p dest=~s state=~p "
          "err=gen_tcp data=~p sock=~p duration=~s",
          log_info(State, [StateName, closed, Sock, duration(State)])),
    {next_state, disconnected, reconnect(tcp_bad(State))};
handle_info(Info, StateName, State) ->
    ?MODULE:StateName(Info, State).

%% @private
terminate(_Reason, _StateName, _State) ->
    ok.

%% @private
code_change(_OldVsn, StateName, State, _Extra) ->
    {ok, StateName, State}.

%% ------------------------------------------------------------------
%% Internal Function Definitions
%% ------------------------------------------------------------------

%% @private
connect(#state{sock = undefined, host=Host, port=Port})
    when is_integer(Port), 0 < Port, Port =< 65535 ->
    SendTimeoutS = logplex_app:config(tcp_syslog_send_timeout_secs),
    HostS = case Host of
                B when is_binary(B) -> binary_to_list(B);
                L when is_list(L) -> L;
                T when is_tuple(T) -> T;
                A when is_atom(A) -> A
{'fraction_non_alphanumeric': 0.23454735933937265, 'fraction_numerical': 0.003204535650459111, 'mean_word_length': 1.9591539022611233, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 70, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2865318', 'n_tokens_mistral': 4928, 'n_tokens_neox': 4798, 'n_words': 1031}