b0VIM 8.0HXYS[Ytkyshmtkyshm-arch.com~tkyshm/erlang/src/efluentc/src/efluentc_client.erlutf-8 3210#"! UtpkmRxa=(ad#k}ypGx^D/  f ( d c % m l [ + 3  r n m K J B u74Z5&% m* pF>wj9 D80%%% Called (1) whenever sys:get_status/1,2 is called by gen_statem or%% @doc%% @private%%--------------------------------------------------------------------callback_mode() -> state_functions. {ok, closed, State, [{next_event, cast, connect}]}. process_flag(trap_exit, true), flush_timeout = Timeout}, flush_buffer_size = BuffSize, port = Port, host = Host, State = #state{id = Id, Timeout = application:get_env(efluentc, flush_timeout, ?DEFAULT_FLUSH_TIMEOUT), BuffSize = application:get_env(efluentc, flush_buffer_size, ?DEFAULT_FLUSH_BUFFER_SIZE), Port = application:get_env(efluentc, port, 24224), Host = application:get_env(efluentc, host, localhost),init([Id]) ->%%--------------------------------------------------------------------%% @end%% {stop, StopReason}%% ignore |%% {CallbackMode, StateName, State, Actions} |%% @spec init(Args) -> {CallbackMode, StateName, State} |%%%% process to initialize.%% gen_statem:start_link/[3,4], this function is called by the new%% Whenever a gen_statem is started using gen_statem:start/[3,4] or%% @doc%% @private%%--------------------------------------------------------------------%%% gen_statem callbacks {ok, Pid}. ets:insert(efluentc, {Id, Pid}), {ok, Pid} = gen_statem:start_link(?MODULE, [Id], []),start_link(Id) ->%%--------------------------------------------------------------------%% @end%% @spec start_link(Id) -> {ok, Pid} | ignore | {error, Error}%%%% function does not return until Module:init/1 has returned.%% initialize. To ensure a synchronized start-up procedure, this%% Creates a gen_statem process which calls Module:init/1 to%% @doc%%--------------------------------------------------------------------%%% API-type client_state() :: #state{}.}). tries = 0 :: integer() sock = undefined :: inet:sock(), flush_timeout = 1000 :: integer(), flush_buffer_size = 0 :: integer(), buffer_caches = [] :: [binary()], % cached buffers when state is closed. buffer = <<>> :: binary(), port = 0 :: integer(), host = localhost :: atom() | inet:ip_address(), id = 0 :: integer(),-record(state, {-define(LOG_FORMAT, "~ts [~ts] ~ts line ~B: ~p~n").-define(DATE_FORMAT, "~4..0w-~2..0w-~2..0wT~2..0w:~2..0w:~2..0w").-define(INFO_LOG(Fmt, Args), infolog(Fmt, Args, ?FILE, ?LINE)).-define(ERR_LOG(Fmt, Args), errlog(Fmt, Args, ?FILE, ?LINE)).-define(BUFFER_CACHE_SIZE, 5). {active, false}]). {packet, 0},-define(GEN_TCP_OPTS, [binary,-define(MAX_RETRY_COUNT, 3). % if post failed, 3 retry-define(DEFAULT_FLUSH_BUFFER_SIZE, 1024 * 10). % defualt 10KB-define(DEFAULT_FLUSH_TIMEOUT, 50). % default 50 msec-define(SERVER, ?MODULE). ]). code_change/4 terminate/3, handle_event/4, empty/3, closed/3, buffered/3, format_status/2, callback_mode/0, init/1,-export([%% gen_statem callbacks-export([start_link/1]).%% API-behaviour(gen_statem).-module(efluentc_client).%%%-------------------------------------------------------------------%%% Created : 2017-11-01 21:35:19.275127%%% @end%%%%%% @doc%%% @copyright (C) 2017, tkyshm%%% @author tkyshm%%%-------------------------------------------------------------------ad'7 =vutik:10% s N )  ^ ] R 1 n Z F 2    w v N  T 7 6 io:format(NewFmt, Args). NewFmt = lists:flatten(io_lib:format(?LOG_FORMAT, [Timestamp, Label, File, Line, Fmt])), Timestamp = lists:flatten(io_lib:format(?DATE_FORMAT, [Y, M, D, HH, MM, SS])), {{Y, M, D}, {HH, MM, SS}} = calendar:now_to_datetime(os:timestamp()),output(Fmt, Args, Label, File, Line) -> output(Fmt, Args, "INFO", File, Line).infolog(Fmt, Args, File, Line) -> output(Fmt, Args, "ERROR", File, Line).errlog(Fmt, Args, File, Line) ->% @privatebackoff(_) -> 5000.backoff(5) -> 3000;backoff(4) -> 3000;backoff(3) -> 3000;backoff(2) -> 2000;backoff(1) -> 2000;backoff(0) -> 1000;% @private gen_tcp:close(Sock). %TODO: safe closesafe_close_sock(Sock) ->% @private [{state_timeout, Timeout, flush}].flush_timeout_action(Timeout) ->% @private [{next_event, cast, flush}].flush_action() ->% @private [{{timeout, connect}, backoff(N), {reconnect, N}}].reconnect_action(N) ->% @private lists:reverse([Buf|lists:reverse(Caches)]).append_buffer_caches(Buf, Caches) -> append_buffer_caches(Buf, Rest); % TODO: save spilled Old caches to file to send all buffered data after reconnectappend_buffer_caches(Buf, Caches = [_Old|Rest]) when length(Caches) > ?BUFFER_CACHE_SIZE ->% @private end. {ok, State#state{buffer = NewBuffs}} NewBuffs -> {buffer_overflow, State#state{buffer = NewBuffs}}; NewBuffs when FBS < byte_size(NewBuffs) -> case <> ofappend_buffer(Msg, State = #state{buffer = Buffs, flush_buffer_size = FBS}) ->% @private end. ok -> send(Sock, Buff, Rest) {error, Reason} -> {error, Reason, Caches}; case gen_tcp:send(Sock, [CBuff]) ofad e8O p [ K 9   T )  | F  s O  { n T 5ji]UUgfZRmR QUL {buffer_overflow, NewState} -> case append_buffer(Msg, State) of Msg = encode_msg(Tag, Data),buffered(cast, {post, Tag, Data}, State = #state{flush_timeout = Timeout}) -> end; {next_state, NextState, NewState} {ok, NextState, NewState} -> {next_state, NextState, NewState, reconnect_action(0)}; ?ERR_LOG("send failed: ~p", [Reason]), {{error, Reason}, NextState, NewState} -> {next_state, NextState, NewState}; ?ERR_LOG("send timeout", []), {{error, timeout}, NextState, NewState} -> case buffered_send(State) ofbuffered(_Event, flush, State) -> {next_state, closed, State#state{tries = 0}}; safe_close_sock(Sock),buffered(_Event, flush, State = #state{sock = Sock, tries = ?MAX_RETRY_COUNT}) ->%% @end%% But theses messages is sent after flush_timeout msec.%% Empty buffered is ready to send messages to fluentd (via efluentc:post/2 api).%% @doc%% @private {next_state, closed, State#state{sock = undefined}, reconnect_action(0)}. safe_close_sock(Sock),empty(info, {tcp_closed, Sock}, State) -> {next_state, buffered, State#state{buffer = NewBuffs}, flush_timeout_action(Timeout)}; NewBuffs = <>, Msg = encode_msg(Tag, Data),empty(cast, {post, Tag, Data}, State = #state{buffer = Buffs, flush_timeout = Timeout}) ->%% @end%% Empty state is ready to send messages to fluentd (via efluentc:post/2 api).%% @doc%% @private {next_state, closed, State#state{buffer = NewBuffs}}. NewBuffs = <>, Msg = encode_msg(Tag, Data),closed(cast, {post, Tag, Data}, #state{buffer = Buffs} = State) -> {next_state, closed, State#state{buffer = Msg, buffer_caches = NewCaches}}; NewCaches = append_buffer_caches(Buffs, Caches), Msg = encode_msg(Tag, Data), when FBS < byte_size(Buffs) -> #state{buffer = Buffs, buffer_caches = Caches, flush_buffer_size = FBS} = State) {post, Tag, Data},closed(cast, end; {next_state, closed, State, reconnect_action(0)} ?ERR_LOG("connect failed: ~p", [Reason]), {error, Reason} -> {next_state, bufferd, NewState, flush_action()}; {ok, NewState, buffered} -> {next_state, empty, NewState}; {ok, NewState, empty} -> case try_connect(State) ofclosed(cast, connect, State) -> end; {next_state, closed, State, reconnect_action(N+1)} ?ERR_LOG("connect failed: ~p", [Reason]), {error, Reason} -> {next_state, buffered, NewState, flush_action()}; ?INFO_LOG("reconnect success", []), {ok, NewState, buffered} -> {next_state, empty, NewState}; ?INFO_LOG("reconnect success", []), {ok, NewState, empty} -> end. end. end. [{data, [{"State", SD}]}] normal -> end. [{data, [{"State", SD}]}] normal -> SD; terminate -> case Opt of {flush_timeout, Data#state.flush_timeout}]}, {flush_buffer_size, Data#state.flush_buffer_size}, {buffer_caches_size, length(Data#state.buffer_caches)}, {buffer_size, byte_size(Data#state.buffer)}, {tries, Data#state.tries}, {port, Data#state.port}, {host, Data#state.host}, SD = {State, [{id, Data#state.id},format_status(Opt, [_PDict, State, Data]) ->%%--------------------------------------------------------------------%% @end%% This callback is optional.%% (2) when gen_statem terminates abnormally.ada\S& D { 8  C  c [ Z N F   s W V K Z / VYXMY" cZYNf<lF_VU=send(Sock, Buff, Caches = [CBuff|Rest]) -> end; ok -> ok {error, Reason} -> {error, Reason, []}; case gen_tcp:send(Sock, [Buff]) ofsend(Sock, Buff, []) -> end. {ok, empty, State#state{buffer = <<>>}} ok -> {{error, Reason}, closed, NewState}; safe_close_sock(Sock), NewState = State#state{sock = undefined, buffer_caches = NewCaches}, {error, Reason, NewCaches} -> {{error, timeout}, buffered, NewState}; NewState = State#state{tries = Tries+1, buffer_caches = NewCaches}, {error, timeout, NewCaches} -> case send(Sock, Buff, Caches) of inet:setopts(Sock, [{active, once}]),buffered_send(State = #state{sock = Sock, buffer = Buff, buffer_caches = Caches, tries = Tries}) ->-spec buffered_send(State :: client_state()) -> {{error, Reason :: term()}, atom(), client_state()} | {ok, atom(), client_state()}.% @private end. {error, Reason} {error, Reason} -> {ok, State#state{sock = Sock}, NextStateName}; gen_tcp:controlling_process(Sock, self()), {ok, Sock} -> case gen_tcp:connect(Host, Port, ?GEN_TCP_OPTS) oftry_connect(State = #state{host = Host, port = Port}, NextStateName) ->% @private try_connect(State, empty).try_connect(State) -> try_connect(State, buffered);try_connect(State = #state{buffer = Buffs}) when byte_size(Buffs) > 0 ->% @private encode_msg(atom_to_binary(Tag, latin1), jiffy:encode(Data)).encode_msg(Tag, Data) when is_atom(Tag) and is_map(Data) -> encode_msg(atom_to_binary(Tag, latin1), list_to_binary(Data));encode_msg(Tag, Data) when is_atom(Tag) and is_list(Data) -> encode_msg(atom_to_binary(Tag, latin1), Data);encode_msg(Tag, Data) when is_atom(Tag) and is_binary(Data) -> encode_msg(Tag, jiffy:encode(Data));encode_msg(Tag, Data) when is_binary(Tag) and is_map(Data) -> encode_msg(Tag, list_to_binary(Data));encode_msg(Tag, Data) when is_binary(Tag) and is_list(Data) -> msgpack:pack(Package, []); Package = [Tag, Msec*1000000+Sec, Data], {Msec, Sec, _} = os:timestamp(),encode_msg(Tag, Data) when is_binary(Tag) and is_binary(Data) ->% @private {ok, StateName, State}.code_change(_OldVsn, StateName, State, _Extra) ->%% @end%% {ok, StateName, NewState}%% @spec code_change(OldVsn, StateName, State, Extra) ->%%%% Convert process state when code is changed%% @doc%% @private ok. ?ERR_LOG("terminated: ~p", [Reason]),terminate(Reason, _StateName, _State) -> ok; safe_close_sock(Sock), ?ERR_LOG("terminated: ~p", [Reason]), ?ERR_LOG("VANISHED BUFFERS!: ~B byte", [byte_size(Buffs)]),terminate(Reason, _StateName, #state{sock = Sock, buffer = Buffs}) when Sock =/= undefined ->%% @end%% @doc%% @private {next_state, flush, State}.handle_event(_EventType, _Content, _StateName, State) -> {next_state, closed, State}; ?ERR_LOG("efluentc id=~w client closed[~w]", [Id, S, self()]),handle_event(_EventType, {tcp_closed, S}, _State, State = #state{id = Id}) -> {next_state, closed, State}; safe_close_sock(Sock), ?ERR_LOG("send failed: pid=~p, reason=~p", [Pid, Reason]),handle_event(_EventType, {'EXIT', Pid, Reason}, _StateName, State = #state{sock = Sock}) ->%% @end%% @doc%% @private {next_state, closed, State#state{sock = undefined}, reconnect_action(0)}. safe_close_sock(Sock),buffered(info, {tcp_closed, Sock}, State) -> end; {next_state, buffered, NewState, flush_timeout_action(Timeout)} {ok, NewState} -> {next_state, buffered, NewState, flush_action()};adR_'qVN ~ Z * \ S 3  g L  Q 0  G & [SO% UuS2tAS2  {buff {buffer_overflow, NewState} -> case append_buffer(Msg, State) of Msg = encode_msg(Tag, Data),buffered(cast, {post, Tag, Data}, State = #state{flush_timeout = Timeout}) -> end; {next_state, NextState, NewState} {ok, NextState, NewState} -> {next_state, NextState, NewState, reconnect_action(0)}; ?ERR_LOG("send failed: ~p", [Reason]), {{error, Reason}, NextState, NewState} -> {next_state, NextState, NewState}; ?ERR_LOG("send timeout", []), {{error, timeout}, NextState, NewState} -> case buffered_send(State) ofbuffered(_Event, flush, State) -> {next_state, closed, State#state{tries = 0}}; safe_close_sock(Sock),buffered(_Event, flush, State = #state{sock = Sock, tries = ?MAX_RETRY_COUNT}) ->%% @end%% But theses messages is sent after flush_timeout msec.%% Empty buffered is ready to send messages to fluentd (via efluentc:post/2 api).%% @doc%% @private {next_state, closed, State#state{sock = undefined}, reconnect_action(0)}. safe_close_sock(Sock),empty(info, {tcp_closed, Sock}, State) -> {next_state, buffered, State#state{buffer = NewBuffs}, flush_timeout_action(Timeout)}; NewBuffs = <>, Msg = encode_msg(Tag, Data),empty(cast, {post, Tag, Data}, State = #state{buffer = Buffs, flush_timeout = Timeout}) ->%% @end%% Empty state is ready to send messages to fluentd (via efluentc:post/2 api).%% @doc%% @private {next_state, closed, State#state{buffer = NewBuffs}}. NewBuffs = <>, Msg = encode_msg(Tag, Data),closed(cast, {post, Tag, Data}, #state{buffer = Buffs} = State) -> {next_state, closed, State#state{buffer = Msg, buffer_caches = NewCaches}}; NewCaches = append_buffer_caches(Buffs, Caches), Msg = encode_msg(Tag, Data), when FBS < byte_size(Buffs) -> #state{buffer = Buffs, buffer_caches = Caches, flush_buffer_size = FBS} = State) {post, Tag, Data},closed(cast, end; {next_state, closed, State, reconnect_action(0)} ?ERR_LOG("connect failed: ~p", [Reason]), {error, Reason} -> {next_state, bufferd, NewState, flush_action()}; {ok, NewState, buffered} -> {next_state, empty, NewState}; {ok, NewState, empty} -> case try_connect(State) ofclosed(cast, connect, State) -> end; {next_state, closed, State, reconnect_action(N+1)} ?ERR_LOG("connect failed: ~p", [Reason]), {error, Reason} -> {next_state, buffered, NewState, flush_action()}; ?INFO_LOG("reconnect success", []), {ok, NewState, buffered} -> {next_state, empty, NewState}; ?INFO_LOG("reconnect success", []), {ok, NewState, empty} -> case try_connect(State) ofclosed({timeout, connect}, {reconnect, N}, State) ->%% @end%% by exponential backoff.%% If process failed to establish connection, do continue to reconnect%%%% so we can adjust it with flush_cache_size.%% Cached size limit is BUFFER_CACHE_SIZE * flush_cache_size,%% connection to fluentd, it flush cached all messages.%% cached in process state record. And when the process established%% Closed state also can be recieved any messages, but theses messages%% @doc%% @private