<commit_before>%%%-------------------------------------------------------------------
%% @copyright <NAME>
%% @author <NAME> <<EMAIL>>
%% @version {@vsn}, {@date} {@time}
%% @doc
%% @end
%%%-------------------------------------------------------------------
-module(logplex_tcpsyslog_drain).

-behaviour(gen_server).

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

%% API
-export([start_link/5
         ,shutdown/1
        ]).

%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
         terminate/2, code_change/3]).

-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
                tref = undefined :: 'undefined' | reference()
               }).

-define(RECONNECT_MSG, reconnect).

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

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

shutdown(Pid) ->
    gen_server:cast(Pid, shutdown),
    ok.

%%====================================================================
%% gen_server callbacks
%%====================================================================

%% @private
init([State0 = #state{drain_id=DrainId, channel_id=ChannelId,
                      host=H, port=P}])
  when H =/= undefined, is_integer(P) ->
    try
        gproc:add_local_name({drain, DrainId}),
        %% This is ugly, but there's no other obvious way to do it.
        gproc:add_local_property({channel, ChannelId}, true),
        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, State, hibernate}
    catch
        error:badarg -> ignore
    end.

%% @private
handle_call(Call, _From, State) ->
    ?WARN("drain_id=~p channel_id=~p dest=~s err=unexpected_call data=~p",
          log_info(State, [Call])),
    {noreply, State}.

%% @private
handle_cast(shutdown, State) ->
    {stop, normal, State};

handle_cast(Msg, State) ->
    ?WARN("drain_id=~p channel_id=~p dest=~s err=unexpected_cast data=~p",
          log_info(State, [Msg])),
    {noreply, State}.

%% @private
handle_info({post, Msg}, State = #state{sock = undefined,
                                        buf = Buf})
  when is_tuple(Msg) ->
    msg_stat(drain_buffered, 1, State),
    NewBuf = logplex_drain_buffer:push(Msg, Buf),
    NewState = State#state{buf=NewBuf},
    {noreply, maybe_reconnect(NewState)};

handle_info({post, Msg}, State = #state{drain_tok = DrainTok,
                                        sock = S})
  when is_tuple(Msg) ->
    case post(Msg, S, DrainTok) of
        ok ->
            msg_stat(drain_delivered, 1, State),
            {noreply, tcp_good(State)};
        {error, Reason} ->
            ?ERR("drain_id=~p channel_id=~p dest=~s at=post err=gen_tcp data=~p",
                 log_info(State, [Reason])),
            {noreply, reconnect(tcp_error, State)}
    end;

handle_info(?RECONNECT_MSG, State = #state{sock = undefined}) ->
    State1 = State#state{tref = undefined},
    case connect(State1) of
        {ok, Sock} ->
            ?INFO("drain_id=~p channel_id=~p dest=~s at=connect try=~p",
                  log_info(State1, [State1#state.failures + 1])),
            NewState = tcp_good(State1#state{sock=Sock}),
            {noreply, post_buffer(NewState)};
        {error, Reason} ->
            NewState = tcp_bad(State1),
            ?ERR("drain_id=~p channel_id=~p dest=~s at=connect "
                 "err=gen_tcp data=~p try=~p last_success=~s",<commit_msg>Buffer messages on send failure.<commit_after>%%%-------------------------------------------------------------------
%% @copyright <NAME>
%% @author <NAME> <<EMAIL>>
%% @version {@vsn}, {@date} {@time}
%% @doc
%% @end
%%%-------------------------------------------------------------------
-module(logplex_tcpsyslog_drain).

-behaviour(gen_server).

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

%% API
-export([start_link/5
         ,shutdown/1
        ]).

%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
         terminate/2, code_change/3]).

-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
                tref = undefined :: 'undefined' | reference()
               }).

-define(RECONNECT_MSG, reconnect).

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

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

shutdown(Pid) ->
    gen_server:cast(Pid, shutdown),
    ok.

%%====================================================================
%% gen_server callbacks
%%====================================================================

%% @private
init([State0 = #state{drain_id=DrainId, channel_id=ChannelId,
                      host=H, port=P}])
  when H =/= undefined, is_integer(P) ->
    try
        gproc:add_local_name({drain, DrainId}),
        %% This is ugly, but there's no other obvious way to do it.
        gproc:add_local_property({channel, ChannelId}, true),
        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, State, hibernate}
    catch
        error:badarg -> ignore
    end.

%% @private
handle_call(Call, _From, State) ->
    ?WARN("drain_id=~p channel_id=~p dest=~s err=unexpected_call data=~p",
          log_info(State, [Call])),
    {noreply, State}.

%% @private
handle_cast(shutdown, State) ->
    {stop, normal, State};

handle_cast(Msg, State) ->
    ?WARN("drain_id=~p channel_id=~p dest=~s err=unexpected_cast data=~p",
          log_info(State, [Msg])),
    {noreply, State}.

%% @private
handle_info({post, Msg}, State = #state{sock = undefined,
                                        buf = Buf})
  when is_tuple(Msg) ->
    msg_stat(drain_buffered, 1, State),
    NewBuf = logplex_drain_buffer:push(Msg, Buf),
    NewState = State#state{buf=NewBuf},
    {noreply, maybe_reconnect(NewState)};

handle_info({post, Msg}, State = #state{drain_tok = DrainTok,
                                        sock = S})
  when is_tuple(Msg) ->
    case post(Msg, S, DrainTok) of
        ok ->
            msg_stat(drain_delivered, 1, State),
            {noreply, tcp_good(State)};
        {error, Reason} ->
            msg_stat(drain_buffered, 1, State),
            NewBuf = logplex_drain_buffer:push(Msg, State#state.buf),
            NewState = State#state{buf=NewBuf},
            ?ERR("drain_id=~p channel_id=~p dest=~s at=post err=gen_tcp data=~p",
                 log_info(NewState, [Reason])),
            {noreply, reconnect(tcp_error, NewState)}
    end;

handle_info(?RECONNECT_MSG, State = #state{sock = undefined}) ->
    State1 = State#state{tref = undefined},
    case connect(State1) of
        {ok, Sock} ->
            ?INFO("drain_id=~p channel_id=~p dest=~s at=connect try=~p",
                  log_info(State1, [State1#state.failures + 1])),
            NewState = tcp_good(State1#state{sock=Sock}),
            {noreply, post_buffer(NewState)};
        {error, Reason} ->
            NewState = tcp_bad(State1),
            ?ERR("drain_id=~p channel_id=~p dest=~s at=connect "
                 "err=gen_tcp data=~p try=~p last_success=~s",
{'fraction_non_alphanumeric': 0.21577798565637668, 'fraction_numerical': 0.00426151127741399, 'mean_word_length': 2.2408218255304817, 'pattern_counts': {'":': 0, '<': 11, '<?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': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3690466', 'n_tokens_mistral': 2979, 'n_tokens_neox': 2838, 'n_words': 537}