diff --git a/src/endpoint/endpoint_kafka.erl b/src/endpoint/endpoint_kafka.erl index ef7bae1..35264c5 100644 --- a/src/endpoint/endpoint_kafka.erl +++ b/src/endpoint/endpoint_kafka.erl @@ -1,4 +1,3 @@ - %%%------------------------------------------------------------------- %%% @author aresei %%% @copyright (C) 2023, @@ -10,201 +9,227 @@ -module(endpoint_kafka). -include("endpoint.hrl"). --behaviour(gen_server). +-behaviour(gen_statem). %% API -export([start_link/2]). -%% gen_server callbacks --export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). +%% gen_statem callbacks +-export([callback_mode/0, init/1, terminate/3, code_change/4]). +-export([disconnected/3, connected/3]). %% 消息重发间隔 -define(RETRY_INTERVAL, 5000). --define(DISCONNECTED, disconnected). --define(CONNECTED, connected). - -record(state, { endpoint :: #endpoint{}, buffer :: endpoint_buffer:buffer(), client_id :: atom(), - client_pid :: undefined | pid(), - status = ?DISCONNECTED + client_pid :: undefined | pid() }). +-type kafka_state() :: disconnected | connected. + %%%=================================================================== %%% API %%%=================================================================== -%% @doc Creates a gen_statem process which calls Module:init/1 to -%% initialize. To ensure a synchronized start-up procedure, this -%% function does not return until Module:init/1 has returned. +-spec start_link(LocalName :: atom(), Endpoint :: #endpoint{}) -> + {ok, pid()} | ignore | {error, term()}. start_link(LocalName, Endpoint = #endpoint{}) when is_atom(LocalName) -> - gen_server:start_link({local, LocalName}, ?MODULE, [Endpoint], []). + gen_statem:start_link({local, LocalName}, ?MODULE, [Endpoint], []). %%%=================================================================== %%% gen_statem callbacks %%%=================================================================== -%% @private -%% @doc Whenever a gen_statem is started using gen_statem:start/[3,4] or -%% gen_statem:start_link/[3,4], this function is called by the new -%% process to initialize. +-spec callback_mode() -> state_functions. +callback_mode() -> + state_functions. + +-spec init(term()) -> gen_statem:init_result(kafka_state(), #state{}). init([Endpoint = #endpoint{id = Id, matcher = Matcher}]) -> ok = iot_log:set_metadata(), - endpoint_subscription:subscribe(Matcher, self()), - erlang:process_flag(trap_exit, true), - %% 创建转发器, 避免阻塞当前进程的创建，因此采用了延时初始化的机制 - erlang:start_timer(0, self(), connect), - %% 初始化存储 + ok = endpoint_subscription:subscribe(Matcher, self()), Buffer = endpoint_buffer:new(Endpoint, 10), ClientId = list_to_atom("brod_client:" ++ integer_to_list(Id)), + {ok, disconnected, #state{endpoint = Endpoint, buffer = Buffer, client_id = ClientId}, + [{state_timeout, 0, connect}]}. - {ok, #state{endpoint = Endpoint, buffer = Buffer, status = ?DISCONNECTED, client_id = ClientId}}. - -%% @private -%% @doc Handling call messages --spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()}, - State :: #state{}) -> - {reply, Reply :: term(), NewState :: #state{}} | - {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} | - {noreply, NewState :: #state{}} | - {noreply, NewState :: #state{}, timeout() | hibernate} | - {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} | - {stop, Reason :: term(), NewState :: #state{}}). -handle_call(get_stat, _From, State = #state{buffer = Buffer}) -> - Stat = endpoint_buffer:stat(Buffer), - {reply, {ok, Stat}, State}. - -%% @private -%% @doc Handling cast messages --spec(handle_cast(Request :: term(), State :: #state{}) -> - {noreply, NewState :: #state{}} | - {noreply, NewState :: #state{}, timeout() | hibernate} | - {stop, Reason :: term(), NewState :: #state{}}). -handle_cast({forward, Metric}, State = #state{buffer = Buffer}) -> - NBuffer = endpoint_buffer:append(Metric, Buffer), - {noreply, State#state{buffer = NBuffer}}; -handle_cast({reload, NEndpoint = #endpoint{matcher = NMatcher}}, State = #state{endpoint = #endpoint{matcher = Matcher}, client_id = ClientId}) -> - ensure_subscription(Matcher, NMatcher), - stop_kafka_client(ClientId), - retry_connect(), - {noreply, State#state{endpoint = NEndpoint, client_pid = undefined, status = ?DISCONNECTED}}; -handle_cast(cleanup, State = #state{buffer = Buffer}) -> - endpoint_buffer:cleanup(Buffer), - {noreply, State}. - -%% @private -%% @doc Handling all non call/cast messages --spec(handle_info(Info :: timeout() | term(), State :: #state{}) -> - {noreply, NewState :: #state{}} | - {noreply, NewState :: #state{}, timeout() | hibernate} | - {stop, Reason :: term(), NewState :: #state{}}). -handle_info({timeout, _, connect}, State = #state{buffer = Buffer, status = ?DISCONNECTED, client_id = ClientId, - endpoint = #endpoint{title = Title, config = #kafka_endpoint{sasl_config = SaslConfig, bootstrap_servers = BootstrapServers, topic = Topic}}}) -> - logger:debug("[endpoint_kafka] endpoint: ~p, create postman", [Title]), - - BaseConfig = [ - {reconnect_cool_down_seconds, 5}, - {socket_options, [{keepalive, true}]} - ], - - ClientConfig = case SaslConfig of - {Mechanism, Username, Password} -> - [{sasl, {Mechanism, Username, Password}}|BaseConfig]; - undefined -> - BaseConfig - end, - - case catch brod:start_link_client(BootstrapServers, ClientId, ClientConfig) of - {ok, ClientPid} -> - case brod:start_producer(ClientId, Topic, _ProducerConfig = []) of - ok -> - NBuffer = endpoint_buffer:trigger_next(Buffer), - {noreply, State#state{buffer = NBuffer, client_pid = ClientPid, status = ?CONNECTED}}; - {error, Reason} -> - logger:debug("[endpoint_kafka] start_producer: ~p, get error: ~p", [ClientId, Reason]), - brod:stop_client(ClientId), - retry_connect(), - {noreply, State#state{status = ?DISCONNECTED, client_pid = undefined}} - end; - Error -> - logger:debug("[endpoint_kafka] start_client: ~p, get error: ~p", [ClientId, Error]), - retry_connect(), - {noreply, State#state{status = ?DISCONNECTED, client_pid = undefined}} - end; - -%% 离线时，忽略数据发送逻辑 -handle_info({next_data, _Id, _Tuple}, State = #state{status = ?DISCONNECTED}) -> - {noreply, State}; -%% 发送数据到mqtt服务器 -handle_info({next_data, Id, Metric}, State = #state{status = ?CONNECTED, client_pid = ClientPid, - endpoint = #endpoint{config = #kafka_endpoint{topic = Topic}}, client_id = ClientId}) -> +-spec disconnected(gen_statem:event_type(), term(), #state{}) -> + gen_statem:event_handler_result(kafka_state(), #state{}). +disconnected({call, From}, get_stat, State = #state{buffer = Buffer}) -> + reply_stat(From, Buffer, State); +disconnected(cast, {forward, Metric}, State = #state{buffer = Buffer}) -> + forward_metric(Metric, Buffer, State); +disconnected(cast, cleanup, State = #state{buffer = Buffer}) -> + cleanup_buffer(Buffer, State); +disconnected(cast, {reload, NEndpoint = #endpoint{matcher = NMatcher}}, + State = #state{endpoint = #endpoint{matcher = Matcher}, client_id = ClientId}) -> + reload_endpoint(Matcher, NMatcher, ClientId, NEndpoint, State); +disconnected(state_timeout, connect, State) -> + try_connect(State); +disconnected(info, {next_data, _Id, _Tuple}, State) -> + {keep_state, State}; +disconnected(info, {ack, Id}, State = #state{buffer = Buffer}) -> + ack_buffer(Id, Buffer, State); +disconnected(info, Info, State) -> + unknown_info(Info, disconnected, State); +disconnected(EventType, EventContent, State) -> + unknown_event(EventType, EventContent, disconnected, State). +-spec connected(gen_statem:event_type(), term(), #state{}) -> + gen_statem:event_handler_result(kafka_state(), #state{}). +connected({call, From}, get_stat, State = #state{buffer = Buffer}) -> + reply_stat(From, Buffer, State); +connected(cast, {forward, Metric}, State = #state{buffer = Buffer}) -> + forward_metric(Metric, Buffer, State); +connected(cast, cleanup, State = #state{buffer = Buffer}) -> + cleanup_buffer(Buffer, State); +connected(cast, {reload, NEndpoint = #endpoint{matcher = NMatcher}}, + State = #state{endpoint = #endpoint{matcher = Matcher}, client_id = ClientId}) -> + reload_endpoint(Matcher, NMatcher, ClientId, NEndpoint, State); +connected(info, {next_data, Id, Metric}, + State = #state{ + client_pid = ClientPid, + client_id = ClientId, + endpoint = #endpoint{config = #kafka_endpoint{topic = Topic}} + }) -> ReceiverPid = self(), AckCb = fun(Partition, BaseOffset) -> logger:debug("[endpoint_kafka] ack partion: ~p, offset: ~p", [Partition, BaseOffset]), ReceiverPid ! {ack, Id} - end, + end, case catch brod:produce_cb(ClientPid, Topic, random, <<>>, Metric, AckCb) of {ok, _CallRef} -> - {noreply, State}; + {keep_state, State}; {ok, _CallRef, _ProducerPid} -> - {noreply, State}; + {keep_state, State}; {error, Reason} -> logger:warning("[endpoint_kafka] produce topic: ~p, get error: ~p", [Topic, Reason]), stop_kafka_client(ClientId), - retry_connect(), - {noreply, State#state{client_pid = undefined, status = ?DISCONNECTED}}; + {next_state, disconnected, State#state{client_pid = undefined}, + [{state_timeout, ?RETRY_INTERVAL, connect}]}; {'EXIT', Reason} -> logger:warning("[endpoint_kafka] produce topic: ~p, exit with reason: ~p", [Topic, Reason]), stop_kafka_client(ClientId), - retry_connect(), - {noreply, State#state{client_pid = undefined, status = ?DISCONNECTED}} + {next_state, disconnected, State#state{client_pid = undefined}, + [{state_timeout, ?RETRY_INTERVAL, connect}]} end; - -handle_info({ack, Id}, State = #state{buffer = Buffer}) -> - NBuffer = endpoint_buffer:ack(Id, Buffer), - {noreply, State#state{buffer = NBuffer}}; - -%% postman进程挂掉时，重新建立新的 -handle_info({'EXIT', ClientPid, Reason}, State = #state{client_pid = ClientPid, endpoint = #endpoint{title = Title}}) -> +connected(info, {ack, Id}, State = #state{buffer = Buffer}) -> + ack_buffer(Id, Buffer, State); +connected(info, {'EXIT', ClientPid, Reason}, + State = #state{client_pid = ClientPid, endpoint = #endpoint{title = Title}}) -> logger:warning("[endpoint_kafka] endpoint: ~p, conn pid exit with reason: ~p", [Title, Reason]), - retry_connect(), - {noreply, State#state{client_pid = undefined, status = ?DISCONNECTED}}; + {next_state, disconnected, State#state{client_pid = undefined}, + [{state_timeout, ?RETRY_INTERVAL, connect}]}; +connected(info, Info, State) -> + unknown_info(Info, connected, State); +connected(EventType, EventContent, State) -> + unknown_event(EventType, EventContent, connected, State). -handle_info(Info, State = #state{status = Status}) -> - logger:warning("[endpoint_kafka] unknown message: ~p, status: ~p", [Info, Status]), - {noreply, State}. - -%% @private -%% @doc This function is called by a gen_server 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_server terminates -%% with Reason. The return value is ignored. --spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()), - State :: #state{}) -> term()). -terminate(Reason, #state{endpoint = #endpoint{title = Title}, buffer = Buffer, client_id = ClientId}) -> +-spec terminate(term(), kafka_state(), #state{}) -> term(). +terminate(Reason, _StateName, #state{endpoint = #endpoint{title = Title}, buffer = Buffer, client_id = ClientId}) -> logger:debug("[endpoint_kafka] endpoint: ~p, terminate with reason: ~p", [Title, Reason]), stop_kafka_client(ClientId), endpoint_buffer:cleanup(Buffer), ok. -%% @private -%% @doc Convert process state when code is changed --spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{}, - Extra :: term()) -> - {ok, NewState :: #state{}} | {error, Reason :: term()}). -code_change(_OldVsn, State = #state{}, _Extra) -> - {ok, State}. +-spec code_change(term() | {down, term()}, kafka_state(), #state{}, term()) -> + {ok, kafka_state(), #state{}} | {error, term()}. +code_change(_OldVsn, StateName, State = #state{}, _Extra) -> + {ok, StateName, State}. %%%=================================================================== %%% Internal functions %%%=================================================================== -retry_connect() -> - erlang:start_timer(?RETRY_INTERVAL, self(), connect). +-spec reply_stat(gen_statem:from(), endpoint_buffer:buffer(), #state{}) -> + gen_statem:event_handler_result(kafka_state(), #state{}). +reply_stat(From, Buffer, State) -> + Stat = endpoint_buffer:stat(Buffer), + {keep_state, State, [{reply, From, {ok, Stat}}]}. + +-spec forward_metric(binary(), endpoint_buffer:buffer(), #state{}) -> + gen_statem:event_handler_result(kafka_state(), #state{}). +forward_metric(Metric, Buffer, State) -> + NBuffer = endpoint_buffer:append(Metric, Buffer), + {keep_state, State#state{buffer = NBuffer}}. + +-spec cleanup_buffer(endpoint_buffer:buffer(), #state{}) -> + gen_statem:event_handler_result(kafka_state(), #state{}). +cleanup_buffer(Buffer, _State) -> + endpoint_buffer:cleanup(Buffer), + keep_state_and_data. + +-spec ack_buffer(integer(), endpoint_buffer:buffer(), #state{}) -> + gen_statem:event_handler_result(kafka_state(), #state{}). +ack_buffer(Id, Buffer, State) -> + NBuffer = endpoint_buffer:ack(Id, Buffer), + {keep_state, State#state{buffer = NBuffer}}. + +-spec reload_endpoint(binary(), binary(), atom(), #endpoint{}, #state{}) -> + gen_statem:event_handler_result(kafka_state(), #state{}). +reload_endpoint(Matcher, NMatcher, ClientId, NEndpoint, State = #state{}) -> + ensure_subscription(Matcher, NMatcher), + stop_kafka_client(ClientId), + {next_state, disconnected, State#state{endpoint = NEndpoint, client_pid = undefined}, + [{state_timeout, 0, connect}]}. + +-spec try_connect(#state{}) -> gen_statem:event_handler_result(kafka_state(), #state{}). +try_connect(State = #state{ + buffer = Buffer, + client_id = ClientId, + endpoint = #endpoint{ + title = Title, + config = #kafka_endpoint{ + sasl_config = SaslConfig, + bootstrap_servers = BootstrapServers, + topic = Topic + } + } + }) -> + logger:debug("[endpoint_kafka] endpoint: ~p, create postman", [Title]), + BaseConfig = [ + {reconnect_cool_down_seconds, 5}, + {socket_options, [{keepalive, true}]} + ], + ClientConfig = case SaslConfig of + {Mechanism, Username, Password} -> + [{sasl, {Mechanism, Username, Password}} | BaseConfig]; + undefined -> + BaseConfig + end, + case catch brod:start_link_client(BootstrapServers, ClientId, ClientConfig) of + {ok, ClientPid} -> + case brod:start_producer(ClientId, Topic, []) of + ok -> + NBuffer = endpoint_buffer:trigger_n(Buffer), + {next_state, connected, State#state{buffer = NBuffer, client_pid = ClientPid}}; + {error, Reason} -> + logger:debug("[endpoint_kafka] start_producer: ~p, get error: ~p", [ClientId, Reason]), + stop_kafka_client(ClientId), + {keep_state, State#state{client_pid = undefined}, + [{state_timeout, ?RETRY_INTERVAL, connect}]} + end; + Error -> + logger:debug("[endpoint_kafka] start_client: ~p, get error: ~p", [ClientId, Error]), + {keep_state, State#state{client_pid = undefined}, + [{state_timeout, ?RETRY_INTERVAL, connect}]} + end. + +-spec unknown_info(term(), kafka_state(), #state{}) -> + gen_statem:event_handler_result(kafka_state(), #state{}). +unknown_info(Info, StateName, State) -> + logger:warning("[endpoint_kafka] unknown message: ~p, status: ~p", [Info, StateName]), + {keep_state, State}. + +-spec unknown_event(gen_statem:event_type(), term(), kafka_state(), #state{}) -> + gen_statem:event_handler_result(kafka_state(), #state{}). +unknown_event(EventType, EventContent, StateName, State) -> + logger:warning("[endpoint_kafka] unknown event: ~p, content: ~p, status: ~p", [EventType, EventContent, StateName]), + {keep_state, State}. -spec ensure_subscription(binary(), binary()) -> ok. ensure_subscription(Matcher, Matcher) ->