From 94efb79eaba74cee0870943571d2d0a9702e1a4a Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Sun, 19 Apr 2026 16:49:52 +0800 Subject: [PATCH] fix --- src/endpoint/endpoint_mqtt.erl | 212 ++++++++++++++++++--------------- 1 file changed, 113 insertions(+), 99 deletions(-) diff --git a/src/endpoint/endpoint_mqtt.erl b/src/endpoint/endpoint_mqtt.erl index 05fa168..de6b4aa 100644 --- a/src/endpoint/endpoint_mqtt.erl +++ b/src/endpoint/endpoint_mqtt.erl @@ -15,7 +15,8 @@ -export([start_link/2]). %% gen_statem callbacks --export([callback_mode/0, init/1, handle_event/4, terminate/3, code_change/4]). +-export([callback_mode/0, init/1, terminate/3, code_change/4]). +-export([disconnected/3, connected/3]). %% 消息重发间隔 -define(RETRY_INTERVAL, 15000). @@ -28,15 +29,12 @@ inflight = #{} }). --type mqtt_state() :: disconnected | connecting | connected. +-type mqtt_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) -> @@ -46,14 +44,10 @@ start_link(LocalName, Endpoint = #endpoint{}) when is_atom(LocalName) -> %%% gen_statem callbacks %%%=================================================================== --spec callback_mode() -> handle_event_function. +-spec callback_mode() -> state_functions. callback_mode() -> - handle_event_function. + state_functions. -%% @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 init(term()) -> gen_statem:init_result(mqtt_state(), #state{}). init([Endpoint = #endpoint{matcher = Matcher}]) -> ok = iot_log:set_metadata(), @@ -62,42 +56,35 @@ init([Endpoint = #endpoint{matcher = Matcher}]) -> Buffer = endpoint_buffer:new(Endpoint, 10), {ok, disconnected, #state{endpoint = Endpoint, buffer = Buffer}, [{next_event, internal, connect}]}. --spec handle_event(gen_statem:event_type(), term(), mqtt_state(), #state{}) -> +-spec disconnected(gen_statem:event_type(), term(), #state{}) -> gen_statem:event_handler_result(mqtt_state(), #state{}). -handle_event({call, From}, get_stat, _StateName, State = #state{buffer = Buffer}) -> - Stat = endpoint_buffer:stat(Buffer), - {keep_state, State, [{reply, From, {ok, Stat}}]}; -handle_event(cast, {forward, Metric}, _StateName, State = #state{buffer = Buffer}) -> - NBuffer = endpoint_buffer:append(Metric, Buffer), - {keep_state, State#state{buffer = NBuffer}}; -handle_event(cast, cleanup, _StateName, #state{buffer = Buffer}) -> - endpoint_buffer:cleanup(Buffer), - keep_state_and_data; -handle_event(cast, {reload, NEndpoint = #endpoint{matcher = NMatcher}}, StateName, +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}, conn_pid = ConnPid}) -> - ensure_subscription(Matcher, NMatcher), - stop_mqtt_conn(ConnPid), - NState = State#state{endpoint = NEndpoint, conn_pid = undefined, inflight = #{}}, - {next_state, connecting, NState, maybe_connect_actions(StateName)}; - -handle_event(internal, connect, disconnected, State) -> - {next_state, connecting, State, [{next_event, internal, do_connect}]}; -handle_event(internal, connect, connecting, State) -> + reload_endpoint(Matcher, NMatcher, ConnPid, NEndpoint, State); +disconnected(internal, connect, State) -> {keep_state, State, [{next_event, internal, do_connect}]}; -handle_event(internal, connect, connected, State) -> - {keep_state, State}; -handle_event(internal, do_connect, connecting, - State = #state{buffer = Buffer, - endpoint = #endpoint{ - title = Title, - config = #mqtt_endpoint{ - host = Host, - port = Port, - username = Username, - password = Password, - client_id = ClientId - } - }}) -> +disconnected(state_timeout, connect, State) -> + {keep_state, State, [{next_event, internal, do_connect}]}; +disconnected(internal, do_connect, + State = #state{ + buffer = Buffer, + endpoint = #endpoint{ + title = Title, + config = #mqtt_endpoint{ + host = Host, + port = Port, + username = Username, + password = Password, + client_id = ClientId + } + } + }) -> logger:debug("[endpoint_mqtt] endpoint: ~ts, create postman", [Title]), Opts = [ {owner, self()}, @@ -120,28 +107,41 @@ handle_event(internal, do_connect, connecting, {next_state, connected, State#state{conn_pid = ConnPid, buffer = NBuffer}}; {error, Reason} -> logger:warning("[endpoint_mqtt] connect get error: ~p", [Reason]), - {next_state, disconnected, State#state{conn_pid = undefined, inflight = #{}}, + {keep_state, State#state{conn_pid = undefined, inflight = #{}}, [{state_timeout, ?RETRY_INTERVAL, connect}]} end; - -handle_event(state_timeout, connect, disconnected, State) -> - {next_state, connecting, State, [{next_event, internal, do_connect}]}; -handle_event(state_timeout, connect, connecting, State) -> - {keep_state, State, [{next_event, internal, do_connect}]}; - -%% 离线或连接中时，忽略数据发送逻辑 -handle_event(info, {next_data, _Id, _Tuple}, disconnected, State) -> +disconnected(info, {next_data, _Id, _Tuple}, State) -> {keep_state, State}; -handle_event(info, {next_data, _Id, _Tuple}, connecting, State) -> +disconnected(info, {'EXIT', ConnPid, Reason}, + State = #state{endpoint = #endpoint{title = Title}, conn_pid = ConnPid}) -> + logger:warning("[endpoint_mqtt] endpoint: ~p, conn pid exit with reason: ~p", [Title, Reason]), + {keep_state, State#state{conn_pid = undefined, inflight = #{}}, + [{state_timeout, ?RETRY_INTERVAL, connect}]}; +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(mqtt_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}, conn_pid = ConnPid}) -> + reload_endpoint(Matcher, NMatcher, ConnPid, NEndpoint, State); +connected(internal, connect, State) -> {keep_state, State}; -%% 发送数据到mqtt服务器 -handle_event(info, {next_data, Id, Metric}, connected, - State = #state{ - conn_pid = ConnPid, - buffer = Buffer, - inflight = InFlight, - endpoint = #endpoint{config = #mqtt_endpoint{topic = Topic, qos = Qos}} - }) -> +connected(info, {next_data, Id, Metric}, + State = #state{ + conn_pid = ConnPid, + buffer = Buffer, + inflight = InFlight, + endpoint = #endpoint{config = #mqtt_endpoint{topic = Topic, qos = Qos}} + }) -> logger:debug("[endpoint_mqtt] will publish topic: ~p, metric: ~p, qos: ~p", [Topic, Metric, Qos]), case emqtt:publish(ConnPid, Topic, #{}, Metric, [{qos, Qos}, {retain, true}]) of ok -> @@ -155,17 +155,16 @@ handle_event(info, {next_data, Id, Metric}, connected, {next_state, disconnected, State#state{conn_pid = undefined, inflight = #{}}, [{state_timeout, ?RETRY_INTERVAL, connect}]} end; - -handle_event(info, {disconnected, ReasonCode, Properties}, connected, State = #state{conn_pid = ConnPid}) -> +connected(info, {disconnected, ReasonCode, Properties}, State = #state{conn_pid = ConnPid}) -> logger:debug("[endpoint_mqtt] Recv a DISONNECT packet - ReasonCode: ~p, Properties: ~p", [ReasonCode, Properties]), stop_mqtt_conn(ConnPid), {next_state, disconnected, State#state{conn_pid = undefined, inflight = #{}}, [{state_timeout, ?RETRY_INTERVAL, connect}]}; -handle_event(info, {publish, Message = #{packet_id := _PacketId, payload := Payload}}, connected, State) -> +connected(info, {publish, Message = #{packet_id := _PacketId, payload := Payload}}, State) -> logger:debug("[endpoint_mqtt] Recv a publish packet: ~p, payload: ~p", [Message, Payload]), {keep_state, State}; -handle_event(info, {puback, #{packet_id := PacketId}}, connected, - State = #state{inflight = Inflight, buffer = Buffer}) -> +connected(info, {puback, #{packet_id := PacketId}}, + State = #state{inflight = Inflight, buffer = Buffer}) -> case maps:take(PacketId, Inflight) of {Id, RestInflight} -> NBuffer = endpoint_buffer:ack(Id, Buffer), @@ -173,40 +172,25 @@ handle_event(info, {puback, #{packet_id := PacketId}}, connected, error -> {keep_state, State} end; - -%% postman进程挂掉时，重新建立新的 -handle_event(info, {'EXIT', ConnPid, Reason}, _StateName, - State = #state{endpoint = #endpoint{title = Title}, conn_pid = ConnPid}) -> +connected(info, {'EXIT', ConnPid, Reason}, + State = #state{endpoint = #endpoint{title = Title}, conn_pid = ConnPid}) -> logger:warning("[endpoint_mqtt] endpoint: ~p, conn pid exit with reason: ~p", [Title, Reason]), {next_state, disconnected, State#state{conn_pid = undefined, inflight = #{}}, [{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_event(info, Info, StateName, State) -> - logger:warning("[endpoint_mqtt] unknown message: ~p, status: ~p", [Info, StateName]), - {keep_state, State}; -handle_event(EventType, EventContent, StateName, State) -> - logger:warning("[endpoint_mqtt] unknown event: ~p, content: ~p, status: ~p", [EventType, EventContent, StateName]), - {keep_state, State}. - -%% @private -%% @doc This function is called by a gen_statem 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_statem terminates -%% with Reason. The return value is ignored. --spec terminate(Reason :: term(), StateName :: mqtt_state(), State :: #state{}) -> term(). +-spec terminate(term(), mqtt_state(), #state{}) -> term(). terminate(Reason, _StateName, #state{endpoint = #endpoint{title = Title}, buffer = Buffer, conn_pid = ConnPid}) -> logger:debug("[endpoint_mqtt] endpoint: ~p, terminate with reason: ~p", [Title, Reason]), stop_mqtt_conn(ConnPid), endpoint_buffer:cleanup(Buffer), ok. -%% @private -%% @doc Convert process state when code is changed --spec code_change(OldVsn :: term() | {down, term()}, - StateName :: mqtt_state(), - State :: #state{}, - Extra :: term()) -> - {ok, StateName :: mqtt_state(), NewState :: #state{}} | {error, term()}. +-spec code_change(term() | {down, term()}, mqtt_state(), #state{}, term()) -> + {ok, mqtt_state(), #state{}} | {error, term()}. code_change(_OldVsn, StateName, State = #state{}, _Extra) -> {ok, StateName, State}. @@ -214,6 +198,44 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) -> %%% Internal functions %%%=================================================================== +-spec reply_stat(gen_statem:from(), endpoint_buffer:buffer(), #state{}) -> + gen_statem:event_handler_result(mqtt_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(mqtt_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(mqtt_state(), #state{}). +cleanup_buffer(Buffer, _State) -> + endpoint_buffer:cleanup(Buffer), + keep_state_and_data. + +-spec reload_endpoint(binary(), binary(), undefined | pid(), #endpoint{}, #state{}) -> + gen_statem:event_handler_result(mqtt_state(), #state{}). +reload_endpoint(Matcher, NMatcher, ConnPid, NEndpoint, State = #state{}) -> + ensure_subscription(Matcher, NMatcher), + stop_mqtt_conn(ConnPid), + {next_state, disconnected, State#state{endpoint = NEndpoint, conn_pid = undefined, inflight = #{}}, + [{next_event, internal, do_connect}]}. + +-spec unknown_info(term(), mqtt_state(), #state{}) -> + gen_statem:event_handler_result(mqtt_state(), #state{}). +unknown_info(Info, StateName, State) -> + logger:warning("[endpoint_mqtt] unknown message: ~p, status: ~p", [Info, StateName]), + {keep_state, State}. + +-spec unknown_event(gen_statem:event_type(), term(), mqtt_state(), #state{}) -> + gen_statem:event_handler_result(mqtt_state(), #state{}). +unknown_event(EventType, EventContent, StateName, State) -> + logger:warning("[endpoint_mqtt] unknown event: ~p, content: ~p, status: ~p", [EventType, EventContent, StateName]), + {keep_state, State}. + -spec ensure_subscription(binary(), binary()) -> ok. ensure_subscription(Matcher, Matcher) -> ok; @@ -221,14 +243,6 @@ ensure_subscription(Matcher, NMatcher) -> ok = endpoint_subscription:unsubscribe(Matcher, self()), endpoint_subscription:subscribe(NMatcher, self()). --spec maybe_connect_actions(mqtt_state()) -> [gen_statem:action()]. -maybe_connect_actions(connected) -> - [{next_event, internal, do_connect}]; -maybe_connect_actions(connecting) -> - [{next_event, internal, do_connect}]; -maybe_connect_actions(disconnected) -> - [{next_event, internal, do_connect}]. - -spec connect_mqtt(list()) -> {ok, pid()} | {error, term()}. connect_mqtt(Opts) -> try