anlicheng/iot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix ws_channel

Browse Source
This commit is contained in:
anlicheng 2023-09-19 14:14:39 +08:00
parent f4cbcf063e
commit f39f92e978
2 changed files with 21 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
apps/iot/src/iot_host.erl
View File

@ -83,7 +83,7 @@ activate(Pid, Auth) when is_pid(Pid), is_boolean(Auth) ->
get_metric(Pid) when is_pid(Pid) ->
gen_server:call(Pid, get_metric).
-spec attach_channel(pid(), pid()) -> ok.
-spec attach_channel(pid(), pid()) -> ok | {error, Reason :: binary()}.
attach_channel(Pid, ChannelPid) when is_pid(Pid), is_pid(ChannelPid) ->
gen_server:call(Pid, {attach_channel, ChannelPid}).
@ -259,18 +259,16 @@ handle_call({activate, Auth}, _From, State = #state{host_id = HostId, uuid = UUI
{reply, ok, State#state{channel_pid = undefined, has_session = false}};
%% 绑定channel
handle_call({attach_channel, ChannelPid}, _From, State = #state{uuid = UUID, channel_pid = undefined}) ->
handle_call({attach_channel, ChannelPid}, _From, State = #state{uuid = UUID, channel_pid = OldChannelPid}) ->
lager:debug("[iot_host] attach_channel host_id uuid: ~p, channel: ~p", [UUID, ChannelPid]),
erlang:monitor(process, ChannelPid),
{reply, ok, State#state{channel_pid = ChannelPid}};
handle_call({attach_channel, ChannelPid}, _From, State = #state{uuid = UUID, channel_pid = OldChannelPid}) when is_pid(OldChannelPid) ->
lager:debug("[iot_host] attach_channel host_id uuid: ~p, old channel: ~p replace with: ~p", [UUID, OldChannelPid, ChannelPid]),
ws_channel:stop(OldChannelPid, closed),
%% 建立到新的channel的monitor
case OldChannelPid =:= undefined of
true ->
erlang:monitor(process, ChannelPid),
{reply, ok, State#state{channel_pid = ChannelPid}};
false ->
lager:debug("[iot_host] attach_channel host_id uuid: ~p, old channel exists: ~p", [UUID, OldChannelPid]),
{reply, {error, <<"channel existed">>}, State#state{channel_pid = ChannelPid}}
end;
%% 授权通过后，才能将主机的状态设置为在线状态
handle_call({create_session, PubKey}, _From, State = #state{uuid = UUID, aes = Aes}) ->
@ -429,8 +427,7 @@ handle_cast({handle, {event, Event0}}, State = #state{uuid = UUID, aes = AES, ha
{noreply, State};
%% 心跳机制
handle_cast(heartbeat, State = #state{uuid = UUID, heartbeat_counter = HeartbeatCounter}) ->
lager:debug("[iot_host] uuip: ~p, get heartbeat, counter is: ~p", [UUID, HeartbeatCounter]),
handle_cast(heartbeat, State = #state{heartbeat_counter = HeartbeatCounter}) ->
{noreply, State#state{heartbeat_counter = HeartbeatCounter + 1}}.
-spec handle_info(Info :: timeout | term(), State :: term()) ->

9
apps/iot/src/websocket/ws_channel.erl
View File

@ -64,12 +64,17 @@ websocket_handle({binary, <<?PACKET_REQUEST, PacketId:32, ?METHOD_AUTH:8, Data/b
{ok, _} ->
%% 尝试启动主机的服务进程
{ok, HostPid} = iot_host_sup:ensured_host_started(UUID),
ok = iot_host:attach_channel(HostPid, self()),
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}}
{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]),