diff --git a/apps/iot/src/http_handler/http_host_handler.erl b/apps/iot/src/http_handler/http_host_handler.erl index 1e63077..7502698 100644 --- a/apps/iot/src/http_handler/http_host_handler.erl +++ b/apps/iot/src/http_handler/http_host_handler.erl @@ -104,16 +104,52 @@ handle_request("POST", "/host/publish_command", _, end end; -%% 处理主机的授权, 激活处理 -handle_request("POST", "/host/activate", _, #{<<"uuid">> := UUID}) when is_binary(UUID) -> +%% 处理主机的授权的"激活/取消激活" 处理 +handle_request("POST", "/host/activate", _, #{<<"uuid">> := UUID, <<"auth">> := Auth, <<"timeout">> := Timeout}) when is_binary(UUID), is_boolean(Auth) -> case iot_host_sup:ensured_host_started(UUID) of {error, Reason} -> lager:debug("[host_handler] activate host_id: ~p, failed with reason: ~p", [UUID, Reason]), - {ok, 200, iot_util:json_error(-1, <<"host not found">>)}; + {ok, 200, iot_util:json_error(400, <<"host not found">>)}; {ok, Pid} when is_pid(Pid) -> lager:debug("[host_handler] reload host_id: ~p, success", [UUID]), - ok = iot_host:activate(Pid), - {ok, 200, iot_util:json_data(<<"success">>)} + %% 下发指令 + {ok, Assoc, ReplyTopic} = iot_mqtt_reply_subscriber:make_assoc(UUID), + Reply = case Auth of + true -> + #{<<"auth">> => true, <<"aes">> => iot_host:get_aes(Pid), <<"reply">> => #{<<"topic">> => ReplyTopic, <<"assoc">> => Assoc}}; + false -> + #{<<"auth">> => false, <<"aes">> => <<"">>, <<"reply">> => #{ <<"topic">> => ReplyTopic, <<"assoc">> => Assoc}} + end, + BinReply = jiffy:encode(Reply, [force_utf8]), + + CommandType = 8, + case iot_host:build_command(Pid, CommandType, BinReply) of + {error, Reason} when is_binary(Reason) -> + {ok, 200, iot_util:json_error(401, Reason)}; + {ok, Topic, BinCommand} -> + case iot_mqtt_publisher:publish(Topic, BinCommand, 2) of + {ok, Ref} -> + receive + {ok, Ref, _PacketId} -> + %% 等待主机返回确认消息 + receive + {host_reply, Assoc, #{<<"code">> := 1}} -> + ok = iot_host:activate(Pid), + {ok, 200, iot_util:json_data(<<"success">>)}; + {host_reply, Assoc, #{<<"code">> := 0, <<"message">> := Message}} when is_binary(Message) -> + {ok, 200, iot_util:json_error(401, <<"操作失败: "/utf8, Message/binary>>)} + after Timeout * 1000 -> + {ok, 200, iot_util:json_error(401, <<"操作超时，请重试: "/utf8>>)} + end + after Timeout * 1000 -> + lager:debug("[iot_host_handler] host_id uuid: ~p, publish topic success, but get ack timeout", [UUID]), + {ok, 200, iot_util:json_error(401, <<"命令执行超时, 请重试"/utf8>>)} + end; + {error, Reason} -> + lager:debug("[iot_host] host_id uuid: ~p, publish topic get error: ~p", [UUID, Reason]), + {ok, 200, iot_util:json_error(402, <<"发送命令到mqtt服务失败"/utf8>>)} + end + end end; handle_request(_, Path, _, _) -> diff --git a/apps/iot/src/iot_host.erl b/apps/iot/src/iot_host.erl index 5d6aef2..f0230e5 100644 --- a/apps/iot/src/iot_host.erl +++ b/apps/iot/src/iot_host.erl @@ -18,7 +18,7 @@ %% API -export([start_link/2, get_name/1, get_pid/1, handle/2, reload/1, activate/1]). --export([get_metric/1, build_command/3, downstream_topic/1]). +-export([get_metric/1, build_command/3, downstream_topic/1, upstream_topic/1, get_aes/1]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). @@ -67,6 +67,10 @@ get_name(UUID) when is_binary(UUID) -> downstream_topic(UUID) when is_binary(UUID) -> <<"host/downstream/", UUID/binary>>. +-spec upstream_topic(UUID :: binary()) -> Topic :: binary(). +upstream_topic(UUID) when is_binary(UUID) -> + <<"host/upstream/", UUID/binary>>. + %% 处理消息 -spec handle(Pid :: pid(), Payload :: binary() | map()) -> no_return(). handle(Pid, Payload) when is_pid(Pid), is_binary(Payload); is_map(Payload) -> @@ -77,6 +81,9 @@ handle(Pid, Payload) when is_pid(Pid), is_binary(Payload); is_map(Payload) -> reload(Pid) when is_pid(Pid) -> gen_server:call(Pid, reload). +get_aes(Pid) when is_pid(Pid) -> + gen_server:call(Pid, get_aes). + %% 激活主机 -spec activate(Pid :: pid()) -> ok. activate(Pid) when is_pid(Pid) -> @@ -111,8 +118,9 @@ init([UUID]) -> {ok, #{<<"status">> := Status}} -> %% 启动心跳定时器 erlang:start_timer(?TICKER_INTERVAL, self(), ping_ticker), + Aes = list_to_binary(iot_util:rand_bytes(32)), - {ok, #state{uuid = UUID, is_activated = (Status /= ?HOST_STATUS_INACTIVE), status = Status, has_session = false}}; + {ok, #state{uuid = UUID, aes = Aes, is_activated = (Status /= ?HOST_STATUS_INACTIVE), status = Status, has_session = false}}; undefined -> lager:warning("[iot_host] host uuid: ~p, loaded from mysql failed", [UUID]), ignore @@ -131,6 +139,9 @@ init([UUID]) -> handle_call(get_metric, _From, State = #state{metrics = Metrics}) -> {reply, {ok, Metrics}, State}; +handle_call(get_aes, _From, State = #state{aes = Aes}) -> + {reply, {ok, Aes}, State}; + %% 重新加载主机信息 handle_call(reload, _From, State = #state{uuid = UUID}) -> %% 重新加载主机信息 @@ -241,10 +252,10 @@ handle_message(Payload, State) when is_binary(Payload) -> handle_message(Message, State); %% create_session操作 -handle_message(#{<<"method">> := <<"create_session">>, <<"params">> := #{<<"pub_key">> := PubKey}}, State = #state{is_activated = IsActivated, uuid = UUID}) -> - lager:debug("[iot_host] host_id uuid: ~p, create_session", [UUID]), +handle_message(#{<<"method">> := <<"create_session">>, <<"params">> := #{<<"pub_key">> := PubKey}}, + State = #state{is_activated = IsActivated, uuid = UUID, aes = Aes}) -> - Aes = list_to_binary(iot_util:rand_bytes(32)), + lager:debug("[iot_host] host_id uuid: ~p, create_session", [UUID]), Reply = case IsActivated of true -> #{<<"a">> => true, <<"aes">> => Aes}; diff --git a/apps/iot/src/iot_mqtt_reply_subscriber.erl b/apps/iot/src/iot_mqtt_reply_subscriber.erl new file mode 100644 index 0000000..4be7b2c --- /dev/null +++ b/apps/iot/src/iot_mqtt_reply_subscriber.erl @@ -0,0 +1,156 @@ +%%%------------------------------------------------------------------- +%%% @author aresei +%%% @copyright (C) 2023, +%%% @doc +%%% 1. 需要考虑集群部署的相关问题，上行的数据可能在集群中共享 +%%% 2. host进程不能直接去监听topic，这样涉及到新增和下线的很多问题 +%%% @end +%%% Created : 12. 3月 2023 21:27 +%%%------------------------------------------------------------------- +-module(iot_mqtt_reply_subscriber). +-author("aresei"). +-include("iot.hrl"). + +-behaviour(gen_server). + +%% API +-export([start_link/0]). +-export([make_assoc/1]). + +%% gen_server callbacks +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). + +-define(SERVER, ?MODULE). + +%% 需要订阅的主题信息 +-define(Topic, <<"host/reply">>). + +-record(state, { + conn_pid :: pid(), + %% 关联数据 + assoc_map = #{}, + %% 任务的自增id + increment_id = 1 +}). + +%%%=================================================================== +%%% API +%%%=================================================================== + +make_assoc(UUID) when is_binary(UUID) -> + gen_server:call(?MODULE, {make_assoc, UUID, self()}). + +%% @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(<<"reply-subscriber">>), + {ok, ConnPid} = emqtt:start_link(Opts), + %% 监听和host相关的全部事件 + {ok, _} = emqtt:connect(ConnPid), + lager:debug("[iot_mqtt_reply_subscriber] connect success, pid: ~p", [ConnPid]), + SubscribeResult = emqtt:subscribe(ConnPid, [{?Topic, 1}]), + + lager:debug("[iot_mqtt_reply_subscriber] subscribe topics: ~p, result is: ~p", [?Topic, SubscribeResult]), + + {ok, #state{conn_pid = ConnPid}}. + +%% @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({make_assoc, UUID, ReceivePid}, _From, State = #state{increment_id = IncrementId, assoc_map = AssocMap}) -> + BinIncrementId = erlang:integer_to_binary(IncrementId), + Assoc = <>, + + {reply, {ok, Assoc, ?Topic}, State#state{assoc_map = maps:put(Assoc, ReceivePid, AssocMap)}}. + +%% @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_reply_subscriber] Recv a DISONNECT packet - ReasonCode: ~p, Properties: ~p", [ReasonCode, Properties]), + {stop, disconnected, State}; +handle_info({publish, #{payload := Payload, qos := Qos, topic := ?Topic}}, State = #state{assoc_map = AssocMap}) -> + lager:debug("[iot_mqtt_reply_subscriber] Recv a reply packet: ~p, qos: ~p", [Payload, Qos]), + case catch jiffy:decode(Payload, [return_maps]) of + Msg = #{<<"assoc">> := Assoc} -> + case maps:take(Assoc, AssocMap) of + error -> + {noreply, State}; + {ReceiverPid, NAssocMap} -> + ReceiverPid ! {host_reply, Assoc, Msg}, + {noreply, State#state{assoc_map = NAssocMap}} + end; + _ -> + {noreply, State} + end; + +handle_info({puback, Packet = #{packet_id := _PacketId}}, State = #state{}) -> + lager:debug("[iot_mqtt_reply_subscriber] receive puback packet: ~p", [Packet]), + {noreply, State}; + +handle_info(Info, State = #state{}) -> + lager:debug("[iot_mqtt_reply_subscriber] 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) -> + %% 取消topic的订阅 + {ok, _Props, _ReasonCode} = emqtt:unsubscribe(ConnPid, #{}, ?Topic), + + ok = emqtt:disconnect(ConnPid), + lager:debug("[iot_mqtt_reply_subscriber] terminate with reason: ~p", [Reason]), + ok; +terminate(Reason, _State) -> + lager:debug("[iot_mqtt_reply_subscriber] 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 +%%%=================================================================== \ No newline at end of file diff --git a/apps/iot/src/iot_sup.erl b/apps/iot/src/iot_sup.erl index f048218..348db54 100644 --- a/apps/iot/src/iot_sup.erl +++ b/apps/iot/src/iot_sup.erl @@ -46,6 +46,15 @@ init([]) -> modules => ['iot_mqtt_subscriber'] }, + #{ + id => 'iot_mqtt_reply_subscriber', + start => {'iot_mqtt_reply_subscriber', start_link, []}, + restart => permanent, + shutdown => 2000, + type => worker, + modules => ['iot_mqtt_reply_subscriber'] + }, + #{ id => 'iot_mqtt_sys_subscriber', start => {'iot_mqtt_sys_subscriber', start_link, []},