调整握手
This commit is contained in:
parent
47a16df2c5
commit
df8cc6f108
@ -26,7 +26,6 @@
|
||||
%% 主机端上报数据类型标识
|
||||
%% 建立到websocket的register关系
|
||||
-define(METHOD_AUTH, 16#00).
|
||||
-define(METHOD_CREATE_SESSION, 16#01).
|
||||
|
||||
-define(METHOD_DATA, 16#02).
|
||||
-define(METHOD_PING, 16#03).
|
||||
|
||||
@ -23,7 +23,7 @@
|
||||
-export([start_link/2, get_name/1, get_alias_name/1, get_pid/1, handle/2, activate/2]).
|
||||
-export([get_metric/1, publish_message/4, get_aes/1, get_status/1]).
|
||||
-export([publish_directive/6, send_directive/5]).
|
||||
-export([create_session/2, attach_channel/2]).
|
||||
-export([attach_channel/2]).
|
||||
-export([reload_device/2, delete_device/2, activate_device/3]).
|
||||
-export([heartbeat/1]).
|
||||
|
||||
@ -88,10 +88,6 @@ get_metric(Pid) when is_pid(Pid) ->
|
||||
attach_channel(Pid, ChannelPid) when is_pid(Pid), is_pid(ChannelPid) ->
|
||||
gen_statem:call(Pid, {attach_channel, ChannelPid}).
|
||||
|
||||
-spec create_session(Pid :: pid(), PubKey :: binary()) -> {ok, Reply :: binary()}.
|
||||
create_session(Pid, PubKey) when is_pid(Pid), is_binary(PubKey) ->
|
||||
gen_statem:call(Pid, {create_session, PubKey}).
|
||||
|
||||
%% 这里占用的的调用进程的时间
|
||||
-spec publish_message(Pid :: pid(), CommandType :: integer(), Params :: binary() | {Encrypt :: atom(), Params :: binary()}, Timeout :: integer()) ->
|
||||
ok | {ok, Response :: binary()} | {error, Reason :: any()}.
|
||||
@ -326,30 +322,24 @@ handle_event({call, From}, {activate, false}, _, State = #state{uuid = UUID, cha
|
||||
{next_state, ?STATE_DENIED, State#state{has_session = false}, [{reply, From, ok}]};
|
||||
|
||||
%% 绑定channel
|
||||
handle_event({call, From}, {attach_channel, ChannelPid}, _, State = #state{uuid = UUID, channel_pid = undefined}) ->
|
||||
lager:debug("[iot_host] attach_channel host_id uuid: ~p, channel: ~p", [UUID, ChannelPid]),
|
||||
handle_event({call, From}, {attach_channel, ChannelPid}, ?STATE_ACTIVATED, State = #state{uuid = UUID, channel_pid = undefined}) ->
|
||||
erlang:monitor(process, ChannelPid),
|
||||
{keep_state, State#state{channel_pid = ChannelPid}, [{reply, From, ok}]};
|
||||
%% 更新主机为在线状态
|
||||
{ok, AffectedRow} = host_bo:change_status(UUID, ?HOST_ONLINE),
|
||||
report_event(UUID, ?HOST_ONLINE),
|
||||
lager:debug("[iot_host] host_id(attach_channel) uuid: ~p, will change status, affected_row: ~p", [UUID, AffectedRow]),
|
||||
|
||||
{keep_state, State#state{channel_pid = ChannelPid, has_session = true}, [{reply, From, ok}]};
|
||||
%% 主机没有激活
|
||||
handle_event({call, From}, {attach_channel, ChannelPid}, ?STATE_DENIED, State = #state{uuid = UUID, channel_pid = undefined}) ->
|
||||
lager:notice("[iot_host] attach_channel host_id uuid: ~p, channel: ~p, host inactivated", [UUID, ChannelPid]),
|
||||
erlang:monitor(process, ChannelPid),
|
||||
{keep_state, State, [{reply, From, {error, <<"host inactivated">>}}]};
|
||||
%% 已经绑定了channel
|
||||
handle_event({call, From}, {attach_channel, _}, _, State = #state{uuid = UUID, channel_pid = OldChannelPid}) ->
|
||||
lager:notice("[iot_host] attach_channel host_id uuid: ~p, old channel exists: ~p", [UUID, OldChannelPid]),
|
||||
{keep_state, State, [{reply, From, {error, <<"channel existed">>}}]};
|
||||
|
||||
%% 授权通过后，才能将主机的状态设置为在线状态
|
||||
handle_event({call, From}, {create_session, PubKey}, ?STATE_ACTIVATED, State = #state{uuid = UUID, aes = Aes}) ->
|
||||
Reply = #{<<"a">> => true, <<"aes">> => Aes},
|
||||
EncReply = iot_cipher_rsa:encode(Reply, PubKey),
|
||||
{ok, AffectedRow} = host_bo:change_status(UUID, ?HOST_ONLINE),
|
||||
report_event(UUID, ?HOST_ONLINE),
|
||||
lager:debug("[iot_host] host_id(session) uuid: ~p, create_session, will change status, affected_row: ~p", [UUID, AffectedRow]),
|
||||
{keep_state, State#state{has_session = true}, [{reply, From, {ok, <<10:8, EncReply/binary>>}}]};
|
||||
|
||||
handle_event({call, From}, {create_session, PubKey}, ?STATE_DENIED, State = #state{uuid = UUID}) ->
|
||||
lager:debug("[iot_host] host_id(denied) uuid: ~p, create_session, will not change host status", [UUID]),
|
||||
Reply = #{<<"a">> => false, <<"aes">> => <<"">>},
|
||||
EncReply = iot_cipher_rsa:encode(Reply, PubKey),
|
||||
{keep_state, State#state{has_session = false}, [{reply, From, {ok, <<10:8, EncReply/binary>>}}]};
|
||||
|
||||
%% 重新加载设备信息
|
||||
handle_event({call, From}, {reload_device, DeviceUUID}, _, State) ->
|
||||
case iot_device_sup:ensured_device_started(DeviceUUID) of
|
||||
|
||||
@ -100,8 +100,16 @@ handle_info({tcp, Socket, <<?PACKET_REQUEST, PacketId:32, ?METHOD_AUTH:8, AuthRe
|
||||
Transport:send(Socket, <<?PACKET_RESPONSE, PacketId:32, 0:8, AuthReplyBin/binary>>),
|
||||
|
||||
{noreply, State#state{uuid = UUID, host_pid = HostPid}};
|
||||
{error, Reason} ->
|
||||
lager:debug("[ws_channel] uuid: ~p, attach channel get error: ~p", [UUID, Reason]),
|
||||
{error, Reason} when is_binary(Reason) ->
|
||||
AuthReplyBin = message_pb:encode_msg(#auth_reply{
|
||||
code = -1,
|
||||
message = Reason,
|
||||
repository_url = <<"">>
|
||||
}),
|
||||
Transport:send(Socket, <<?PACKET_RESPONSE, PacketId:32, 0:8, AuthReplyBin/binary>>),
|
||||
|
||||
lager:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]),
|
||||
|
||||
{stop, State}
|
||||
end
|
||||
end;
|
||||
@ -110,10 +118,6 @@ handle_info({tcp, Socket, <<?PACKET_REQUEST, PacketId:32, ?METHOD_AUTH:8, AuthRe
|
||||
{stop, State}
|
||||
end;
|
||||
|
||||
handle_info({binary, <<?PACKET_REQUEST, PacketId:32, ?METHOD_CREATE_SESSION:8, PubKey/binary>>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
{ok, Reply} = iot_host:create_session(HostPid, PubKey),
|
||||
{reply, {binary, <<?PACKET_RESPONSE, PacketId:32, Reply/binary>>}, State};
|
||||
|
||||
handle_info({binary, <<?PACKET_REQUEST, _PacketId:32, ?METHOD_DATA:8, Data/binary>>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
iot_host:handle(HostPid, {data, Data}),
|
||||
{ok, State};
|
||||
|
||||
@ -1,174 +0,0 @@
|
||||
%%%-------------------------------------------------------------------
|
||||
%%% @author licheng5
|
||||
%%% @copyright (C) 2021, <COMPANY>
|
||||
%%% @doc
|
||||
%%%
|
||||
%%% @end
|
||||
%%% Created : 11. 1月 2021 上午12:17
|
||||
%%%-------------------------------------------------------------------
|
||||
-module(ws_channel).
|
||||
-author("licheng5").
|
||||
-include("iot.hrl").
|
||||
|
||||
%% API
|
||||
-export([init/2]).
|
||||
-export([websocket_init/1, websocket_handle/2, websocket_info/2, terminate/3]).
|
||||
-export([publish/3, stop/2, send/2]).
|
||||
|
||||
-record(state, {
|
||||
uuid :: undefined | binary(),
|
||||
%% 用户进程id
|
||||
host_pid = undefined,
|
||||
%% 发送消息对应的id
|
||||
packet_id = 1 :: integer(),
|
||||
|
||||
%% 请求响应的对应关系
|
||||
inflight = #{}
|
||||
}).
|
||||
|
||||
%% 向通道中写入消息
|
||||
-spec publish(Pid :: pid(), ReceiverPid :: pid(), Msg :: binary()) -> Ref :: reference().
|
||||
publish(Pid, ReceiverPid, Msg) when is_pid(Pid), is_binary(Msg) ->
|
||||
Ref = make_ref(),
|
||||
Pid ! {publish, ReceiverPid, Ref, Msg},
|
||||
Ref.
|
||||
|
||||
%% 向通道中写入消息
|
||||
-spec send(Pid :: pid(), Msg :: binary()) -> no_return().
|
||||
send(Pid, Msg) when is_pid(Pid), is_binary(Msg) ->
|
||||
Pid ! {send, Msg}.
|
||||
|
||||
%% 关闭方法
|
||||
-spec stop(Pid :: pid(), Reason :: any()) -> no_return().
|
||||
stop(undefined, _Reason) ->
|
||||
ok;
|
||||
stop(Pid, Reason) when is_pid(Pid) ->
|
||||
Pid ! {stop, Reason}.
|
||||
|
||||
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
|
||||
%% 逻辑处理方法
|
||||
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
|
||||
|
||||
init(Req, Opts) ->
|
||||
{cowboy_websocket, Req, Opts}.
|
||||
|
||||
websocket_init(_State) ->
|
||||
lager:debug("[ws_channel] get a new connection"),
|
||||
%% 初始状态为true
|
||||
{ok, #state{packet_id = 1}}.
|
||||
|
||||
websocket_handle({binary, <<?PACKET_REQUEST, PacketId:32, ?METHOD_AUTH:8, Data/binary>>}, State) ->
|
||||
#{<<"uuid">> := UUID, <<"timestamp">> := Timestamp, <<"salt">> := Salt, <<"username">> := Username, <<"token">> := Token} = jiffy:decode(Data, [return_maps]),
|
||||
lager:debug("[ws_channel] auth uuid: ~p, request message: ~p", [UUID, Data]),
|
||||
case iot_auth:check(Username, Token, UUID, Salt, Timestamp) of
|
||||
true ->
|
||||
case host_bo:get_host_by_uuid(UUID) of
|
||||
undefined ->
|
||||
lager:warning("[ws_channel] uuid: ~p, user: ~p, host not found", [UUID, Username]),
|
||||
{stop, State};
|
||||
{ok, _} ->
|
||||
%% 尝试启动主机的服务进程
|
||||
{ok, HostPid} = iot_host_sup:ensured_host_started(UUID),
|
||||
case iot_host:attach_channel(HostPid, self()) of
|
||||
ok ->
|
||||
%% 建立到host的monitor
|
||||
erlang:monitor(process, HostPid),
|
||||
Reply = jiffy:encode(#{<<"code">> => 1, <<"message">> => <<"ok">>}, [force_utf8]),
|
||||
|
||||
{reply, {binary, <<?PACKET_RESPONSE, PacketId:32, 0:8, Reply/binary>>}, State#state{uuid = UUID, host_pid = HostPid}};
|
||||
{error, Reason} ->
|
||||
lager:debug("[ws_channel] uuid: ~p, attach channel get error: ~p", [UUID, Reason]),
|
||||
{stop, State}
|
||||
end
|
||||
end;
|
||||
false ->
|
||||
lager:warning("[ws_channel] uuid: ~p, user: ~p, auth failed", [UUID, Username]),
|
||||
{stop, State}
|
||||
end;
|
||||
|
||||
websocket_handle({binary, <<?PACKET_REQUEST, PacketId:32, ?METHOD_CREATE_SESSION:8, PubKey/binary>>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
{ok, Reply} = iot_host:create_session(HostPid, PubKey),
|
||||
{reply, {binary, <<?PACKET_RESPONSE, PacketId:32, Reply/binary>>}, State};
|
||||
|
||||
websocket_handle({binary, <<?PACKET_REQUEST, _PacketId:32, ?METHOD_DATA:8, Data/binary>>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
iot_host:handle(HostPid, {data, Data}),
|
||||
{ok, State};
|
||||
|
||||
websocket_handle({binary, <<?PACKET_REQUEST, _PacketId:32, ?METHOD_PING:8, CipherMetric/binary>>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
iot_host:handle(HostPid, {ping, CipherMetric}),
|
||||
{ok, State};
|
||||
|
||||
websocket_handle({binary, <<?PACKET_REQUEST, _PacketId:32, ?METHOD_INFORM:8, CipherInfo/binary>>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
iot_host:handle(HostPid, {inform, CipherInfo}),
|
||||
{ok, State};
|
||||
|
||||
websocket_handle({binary, <<?PACKET_REQUEST, _PacketId:32, ?METHOD_FEEDBACK_STEP:8, CipherInfo/binary>>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
iot_host:handle(HostPid, {feedback_step, CipherInfo}),
|
||||
{ok, State};
|
||||
|
||||
websocket_handle({binary, <<?PACKET_REQUEST, _PacketId:32, ?METHOD_FEEDBACK_RESULT:8, CipherInfo/binary>>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
iot_host:handle(HostPid, {feedback_result, CipherInfo}),
|
||||
{ok, State};
|
||||
|
||||
websocket_handle({binary, <<?PACKET_REQUEST, _PacketId:32, ?METHOD_EVENT:8, CipherEvent/binary>>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
iot_host:handle(HostPid, {event, CipherEvent}),
|
||||
{ok, State};
|
||||
|
||||
websocket_handle({binary, <<?PACKET_REQUEST, _PacketId:32, ?METHOD_AI_EVENT:8, CipherEvent/binary>>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
iot_host:handle(HostPid, {ai_event, CipherEvent}),
|
||||
{ok, State};
|
||||
|
||||
%% 主机端的消息响应
|
||||
websocket_handle({binary, <<?PACKET_PUBLISH_RESPONSE, 0:32, Body/binary>>}, State = #state{uuid = UUID}) ->
|
||||
lager:debug("[ws_channel] uuid: ~p, get send response message: ~p", [UUID, Body]),
|
||||
{ok, State};
|
||||
websocket_handle({binary, <<?PACKET_PUBLISH_RESPONSE, PacketId:32, Body/binary>>}, State = #state{uuid = UUID, inflight = Inflight}) when PacketId > 0 ->
|
||||
lager:debug("[ws_channel] uuid: ~p, get publish response message: ~p, packet_id: ~p", [UUID, Body, PacketId]),
|
||||
case maps:take(PacketId, Inflight) of
|
||||
error ->
|
||||
lager:warning("[ws_channel] get unknown publish response message: ~p, packet_id: ~p", [Body, PacketId]),
|
||||
{ok, State};
|
||||
{{ReceiverPid, Ref}, NInflight} ->
|
||||
case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of
|
||||
true when Body == <<>> ->
|
||||
ReceiverPid ! {ws_response, Ref};
|
||||
true ->
|
||||
ReceiverPid ! {ws_response, Ref, Body};
|
||||
false ->
|
||||
lager:warning("[ws_channel] get publish response message: ~p, packet_id: ~p, but receiver_pid is deaded", [Body, PacketId])
|
||||
end,
|
||||
{ok, State#state{inflight = NInflight}}
|
||||
end;
|
||||
|
||||
websocket_handle(Info, State) ->
|
||||
lager:error("[ws_channel] get a unknown message: ~p, channel will closed", [Info]),
|
||||
{stop, State}.
|
||||
|
||||
%% 处理关闭信号
|
||||
websocket_info({stop, Reason}, State) ->
|
||||
lager:debug("[ws_channel] the channel will be closed with reason: ~p", [Reason]),
|
||||
{stop, State};
|
||||
|
||||
%% 发送消息
|
||||
websocket_info({publish, ReceiverPid, Ref, Msg}, State = #state{packet_id = PacketId, inflight = Inflight}) when is_binary(Msg) ->
|
||||
NInflight = maps:put(PacketId, {ReceiverPid, Ref}, Inflight),
|
||||
{reply, {binary, <<?PACKET_PUBLISH, PacketId:32, Msg/binary>>}, State#state{packet_id = PacketId + 1, inflight = NInflight}};
|
||||
|
||||
%% 发送消息, 不需要等待回复
|
||||
websocket_info({send, Msg}, State) when is_binary(Msg) ->
|
||||
{reply, {binary, <<?PACKET_PUBLISH, 0:32, Msg/binary>>}, State};
|
||||
|
||||
%% 用户进程关闭，则关闭通道
|
||||
websocket_info({'DOWN', _, process, HostPid, Reason}, State = #state{uuid = UUID, host_pid = HostPid}) ->
|
||||
lager:debug("[ws_channel] uuid: ~p, channel will close because host exited with reason: ~p", [UUID, Reason]),
|
||||
{stop, State};
|
||||
|
||||
%% 处理其他未知消息
|
||||
websocket_info(Info, State = #state{uuid = UUID}) ->
|
||||
lager:debug("[ws_channel] channel get unknown info: ~p, uuid: ~p", [Info, UUID]),
|
||||
{ok, State}.
|
||||
|
||||
%% 进程关闭事件
|
||||
terminate(Reason, _Req, State) ->
|
||||
lager:debug("[ws_channel] channel close with reason: ~p, state is: ~p", [Reason, State]),
|
||||
ok.
|
||||
Loading…
x
Reference in New Issue
Block a user