fix

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,32 +56,24 @@ 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}) ->
+disconnected({call, From}, get_stat, State = #state{buffer = Buffer}) ->
-    Stat = endpoint_buffer:stat(Buffer),
+    reply_stat(From, Buffer, State);
-    {keep_state, State, [{reply, From, {ok, Stat}}]};
+disconnected(cast, {forward, Metric}, State = #state{buffer = Buffer}) ->
-handle_event(cast, {forward, Metric}, _StateName, State = #state{buffer = Buffer}) ->
+    forward_metric(Metric, Buffer, State);
-    NBuffer = endpoint_buffer:append(Metric, Buffer),
+disconnected(cast, cleanup, State = #state{buffer = Buffer}) ->
-    {keep_state, State#state{buffer = NBuffer}};
+    cleanup_buffer(Buffer, State);
-handle_event(cast, cleanup, _StateName, #state{buffer = Buffer}) ->
+disconnected(cast, {reload, NEndpoint = #endpoint{matcher = NMatcher}},
-    endpoint_buffer:cleanup(Buffer),
-    keep_state_and_data;
-handle_event(cast, {reload, NEndpoint = #endpoint{matcher = NMatcher}}, StateName,
              State = #state{endpoint = #endpoint{matcher = Matcher}, conn_pid = ConnPid}) ->
-    ensure_subscription(Matcher, NMatcher),
+    reload_endpoint(Matcher, NMatcher, ConnPid, NEndpoint, State);
-    stop_mqtt_conn(ConnPid),
+disconnected(internal, connect, State) ->
-    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) ->
     {keep_state, State, [{next_event, internal, do_connect}]};
-handle_event(internal, connect, connected, State) ->
+disconnected(state_timeout, connect, State) ->
-    {keep_state, State};
+    {keep_state, State, [{next_event, internal, do_connect}]};
-handle_event(internal, do_connect, connecting,
+disconnected(internal, do_connect,
-             State = #state{buffer = Buffer,
+             State = #state{
+                 buffer = Buffer,
                  endpoint = #endpoint{
                      title = Title,
                      config = #mqtt_endpoint{
@@ -97,7 +83,8 @@ handle_event(internal, do_connect, connecting,
                          password = Password,
                          client_id = ClientId
                      }
-                            }}) ->
+                 }
+             }) ->
     logger:debug("[endpoint_mqtt] endpoint: ~ts, create postman", [Title]),
     Opts = [
         {owner, self()},
@@ -120,22 +107,35 @@ 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;
-
+disconnected(info, {next_data, _Id, _Tuple}, State) ->
-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) ->
     {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服务器
+connected(info, {next_data, Id, Metric},
-handle_event(info, {next_data, Id, Metric}, connected,
           State = #state{
               conn_pid = ConnPid,
               buffer = Buffer,
@@ -155,16 +155,15 @@ handle_event(info, {next_data, Id, Metric}, connected,
             {next_state, disconnected, State#state{conn_pid = undefined, inflight = #{}},
              [{state_timeout, ?RETRY_INTERVAL, connect}]}
     end;
-
+connected(info, {disconnected, ReasonCode, Properties}, State = #state{conn_pid = ConnPid}) ->
-handle_event(info, {disconnected, ReasonCode, Properties}, connected, 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,
+connected(info, {puback, #{packet_id := PacketId}},
           State = #state{inflight = Inflight, buffer = Buffer}) ->
     case maps:take(PacketId, Inflight) of
         {Id, RestInflight} ->
@@ -173,40 +172,25 @@ handle_event(info, {puback, #{packet_id := PacketId}}, connected,
         error ->
             {keep_state, State}
     end;
-
+connected(info, {'EXIT', ConnPid, Reason},
-%% postman进程挂掉时，重新建立新的
-handle_event(info, {'EXIT', ConnPid, Reason}, _StateName,
           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) ->
+-spec terminate(term(), mqtt_state(), #state{}) -> term().
-    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().
 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
+-spec code_change(term() | {down, term()}, mqtt_state(), #state{}, term()) ->
-%% @doc Convert process state when code is changed
+    {ok, mqtt_state(), #state{}} | {error, term()}.
--spec code_change(OldVsn :: term() | {down, term()},
-                  StateName :: mqtt_state(),
-                  State :: #state{},
-                  Extra :: term()) ->
-    {ok, StateName :: mqtt_state(), NewState :: #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