fix

apps/iot/src/iot_mqtt_publisher.erl

@@ -0,0 +1,160 @@
+%%%-------------------------------------------------------------------
+%%% @author aresei
+%%% @copyright (C) 2023, <COMPANY>
+%%% @doc
+%%%
+%%% @end
+%%% Created : 12. 3月 2023 21:27
+%%%-------------------------------------------------------------------
+-module(iot_mqtt_publisher).
+-author("aresei").
+-include("iot.hrl").
+
+-behaviour(gen_server).
+
+%% API
+-export([start_link/0, publish/3, publish_result/2]).
+
+%% gen_server callbacks
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
+
+-define(SERVER, ?MODULE).
+
+-record(state, {
+    conn_pid :: pid(),
+    inflight = #{}
+}).
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+
+-spec publish(binary(), binary(), integer()) -> {ok, Ref :: reference()} | {error, term()}.
+publish(Topic, Message, Qos) when is_binary(Topic), is_binary(Message), is_integer(Qos) ->
+    gen_server:call(?MODULE, {publish, self(), Topic, Message, Qos}).
+
+%% 获取publish的结果
+-spec publish_result(tuple(), Timeout :: integer()) -> {ok, PacketId :: integer()} | {error, Reason :: any()}.
+publish_result({error, Reason}, _) ->
+    {error, Reason};
+publish_result({ok, Ref}, Timeout) when is_reference(Ref), is_integer(Timeout) ->
+    receive
+        {ok, Ref, PacketId} ->
+            {ok, PacketId}
+    after Timeout ->
+        {error, timeout}
+    end.
+
+%% @doc Spawns the server and registers the local name (unique)
+-spec(start_link() ->
+    {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
+start_link() ->
+    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+%%%===================================================================
+%%% gen_server callbacks
+%%%===================================================================
+
+%% @private
+%% @doc Initializes the server
+-spec(init(Args :: term()) ->
+    {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
+    {stop, Reason :: term()} | ignore).
+init([]) ->
+    %% 建立到emqx服务器的连接
+    Opts = iot_config:emqt_opts(),
+    case emqtt:start_link(Opts) of
+        {ok, ConnPid} ->
+            lager:debug("[iot_mqtt_publisher] connect success, pid: ~p", [ConnPid]),
+            {ok, #state{conn_pid = ConnPid}};
+        ignore ->
+            lager:debug("[iot_mqtt_publisher] connect emqx get ignore"),
+            {stop, ignore};
+        {error, Reason} ->
+            lager:debug("[iot_mqtt_publisher] connect emqx get error: ~p", [Reason]),
+            {stop, Reason}
+    end.
+
+%% @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({publish, ReceiverPid, Topic, Message, Qos}, _From, State = #state{conn_pid = ConnPid, inflight = InFlight}) ->
+    case emqtt:publish(ConnPid, Topic, #{}, Message, [{qos, Qos}, {retain, true}]) of
+        {ok, PacketId} ->
+            Ref = make_ref(),
+            {reply, {ok, Ref}, State#state{inflight = maps:put(PacketId, {ReceiverPid, Ref, Message}, InFlight)}};
+        {error, Reason} ->
+            {reply, {error, Reason}, State}
+    end.
+
+%% @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(_Request, State = #state{}) ->
+    {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({disconnect, ReasonCode, Properties}, State = #state{}) ->
+    lager:debug("[iot_mqtt_publisher] Recv a DISONNECT packet - ReasonCode: ~p, Properties: ~p", [ReasonCode, Properties]),
+    {stop, disconnected, State};
+handle_info({publish, Message = #{packet_id := _PacketId, payload := Payload}}, State = #state{conn_pid = _ConnPid}) ->
+    lager:debug("[iot_mqtt_publisher] Recv a publish packet: ~p, payload: ~p", [Message, Payload]),
+    {noreply, State};
+handle_info({puback, Packet = #{packet_id := PacketId}}, State = #state{inflight = Inflight}) ->
+    case maps:take(PacketId, Inflight) of
+        {{ReceiverPid, Ref, Message}, Inflight1} ->
+            lager:debug("[iot_mqtt_publisher] receive puback packet: ~p, assoc message: ~p", [Packet, Message]),
+            ReceiverPid ! {ok, Ref, PacketId},
+            {noreply, State#state{inflight = Inflight1}};
+        error ->
+            lager:warning("[iot_mqtt_publisher] receive unknown puback packet: ~p", [Packet]),
+            {noreply, State}
+    end;
+
+handle_info(Info, State = #state{}) ->
+    lager:debug("[iot_mqtt_publisher] get info: ~p", [Info]),
+    {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 = #state{conn_pid = ConnPid}) when is_pid(ConnPid) ->
+    {ok, _Props, _ReasonCode} = emqtt:unsubscribe(ConnPid, #{}, <<"hello">>),
+    ok = emqtt:disconnect(ConnPid),
+    ok = emqtt:stop(ConnPid),
+    lager:debug("[iot_mqtt_publisher] terminate with reason: ~p", [Reason]),
+    ok;
+terminate(Reason, _State) ->
+    lager:debug("[iot_mqtt_publisher] terminate with reason: ~p", [Reason]),
+    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}.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================

apps/iot/src/iot_mqtt_subscriber.erl

@@ -6,7 +6,7 @@
 %%% @end
 %%% Created : 12. 3月 2023 21:27
 %%%-------------------------------------------------------------------
--module(iot_mqtt_share_subscriber).
+-module(iot_mqtt_subscriber).
 -author("aresei").
 -include("iot.hrl").
 
@@ -121,7 +121,7 @@ handle_info({timeout, _, subscribe_ticker}, State = #state{conn_pid = ConnPid})
     %% 监听和host相关的全部事件
     {ok, _} = emqtt:connect(ConnPid),
     Topics = [
-
+        {<<"$share/server/register">>, 1}
     ],
     SubscribeResult = emqtt:subscribe(ConnPid, Topics),
     lager:debug("[iot_mqtt_share_subscriber] subscribe result is: ~p", [SubscribeResult]),

apps/iot/src/iot_sup.erl

@@ -56,12 +56,20 @@ init([]) ->
         },
 
         #{
-            id => 'iot_mqtt_share_subscriber',
+            id => 'iot_mqtt_subscriber',
-            start => {'iot_mqtt_share_subscriber', start_link, []},
+            start => {'iot_mqtt_subscriber', start_link, []},
             restart => permanent,
             shutdown => 2000,
             type => worker,
-            modules => ['iot_mqtt_share_subscriber']
+            modules => ['iot_mqtt_subscriber']
+        },
+        #{
+            id => 'iot_mqtt_publisher',
+            start => {'iot_mqtt_publisher', start_link, []},
+            restart => permanent,
+            shutdown => 2000,
+            type => worker,
+            modules => ['iot_mqtt_publisher']
         }
     ],
     {ok, {SupFlags, ChildSpecs}}.