diff --git a/docs/efka_iot_protocol.md b/docs/efka_iot_protocol.md index 3dcef47..d559ef9 100644 --- a/docs/efka_iot_protocol.md +++ b/docs/efka_iot_protocol.md @@ -47,17 +47,16 @@ }}} ``` -`iot` 回复: +`iot` 鉴权成功时回复: ```erlang {response, Ref, {auth_response, ok}} -{response, Ref, {auth_response, {error, {failed, Reason}}}} ``` 处理语义: - `ok`:`efka` 进入 `activated` 状态。 -- `{error, {failed, Reason}}`:鉴权失败,连接关闭后重连。 +- 鉴权失败、host 不存在、host 未启动或 attach channel 失败时,`iot` 不返回业务错误响应,直接关闭连接;`efka` 通过 socket close/error 或 auth timeout 进入重连流程。 ## 授权控制 diff --git a/src/host/iot_host.erl b/src/host/iot_host.erl index 954fc65..9e254ab 100644 --- a/src/host/iot_host.erl +++ b/src/host/iot_host.erl @@ -22,6 +22,7 @@ %% API -export([start_link/2, get_name/1, get_alias_name/1, get_pid/1, handle/2, activate/2]). +-export([lookup_pid/1]). -export([get_metric/1, get_status/1, kill/1]). %% 通讯相关 -export([pub/4, attach_channel/2]). @@ -52,6 +53,15 @@ get_pid(UUID) when is_binary(UUID) -> Name = get_name(UUID), gproc:whereis_name({n, l, Name}). +-spec lookup_pid(UUID :: binary()) -> {ok, pid()} | {error, Reason :: any()}. +lookup_pid(UUID) when is_binary(UUID) -> + case get_pid(UUID) of + undefined -> + {error, <<"host not found">>}; + HostPid -> + {ok, HostPid} + end. + -spec get_name(UUID :: binary()) -> term(). get_name(UUID) when is_binary(UUID) -> {iot_host, UUID}. diff --git a/src/transport/tcp/ssl_channel.erl b/src/transport/tcp/ssl_channel.erl index 635d0a2..8447114 100644 --- a/src/transport/tcp/ssl_channel.erl +++ b/src/transport/tcp/ssl_channel.erl @@ -176,40 +176,17 @@ handle_request_frame(Ref, {auth_request, #{uuid := UUID}}, State = #state{is_aut logger:warning("[ws_channel] repeated auth request, ref: ~p, uuid: ~p, close channel", [Ref, UUID]), {stop, repeated_auth, State}; handle_request_frame(Ref, {auth_request, #{uuid := UUID, token := Token, timestamp := Timestamp}}, State = #state{transport = Transport, socket = Socket}) -> - logger:debug("[ws_channel] auth uuid: ~p", [UUID]), - case auth(Token, UUID, Timestamp) of - ok -> - case iot_api_client:get_host_by_uuid(UUID) of - undefined -> - logger:warning("[ws_channel] uuid: ~p, token: ~p, host not found", [UUID, Token]), - stop_after_reply(Transport, Socket, Ref, {auth_response, {error, {failed, <<"host not found">>}}}, normal, State); - {ok, _} -> - %% 尝试启动主机的服务进程 - case iot_host_sup:ensured_host_started(UUID) of - {ok, HostPid} -> - case iot_host:attach_channel(HostPid, self()) of - ok -> - erlang:monitor(process, HostPid), - case send_reply_frame(Transport, Socket, Ref, {auth_response, ok}) of - ok -> - {noreply, State#state{uuid = UUID, is_authed = true, host_pid = HostPid}}; - {error, Reason} -> - logger:warning("[ws_channel] uuid: ~p, send auth response failed: ~p", [UUID, Reason]), - {stop, {send_failed, Reason}, State} - end; - {error, Reason} when is_binary(Reason) -> - logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]), - stop_after_reply(Transport, Socket, Ref, {auth_response, {error, {failed, Reason}}}, Reason, State) - end; - {error, Reason} -> - logger:warning("[ws_channel] uuid: ~p, ensure host failed: ~p", [UUID, Reason]), - ReasonBin = reason_to_binary(Reason), - stop_after_reply(Transport, Socket, Ref, {auth_response, {error, {failed, ReasonBin}}}, Reason, State) - end - end; - {error, Reason} -> - logger:warning("[ws_channel] uuid: ~p, token: ~p, auth failed, reason: ~p", [UUID, Token, Reason]), - stop_after_reply(Transport, Socket, Ref, {auth_response, {error, {failed, Reason}}}, Reason, State) + maybe + ok ?= auth(Token, UUID, Timestamp), + {ok, HostPid} ?= iot_host:lookup_pid(UUID), + ok ?= iot_host:attach_channel(HostPid, self()), + erlang:monitor(process, HostPid), + ok ?= send_reply_frame(Transport, Socket, Ref, {auth_response, ok}), + logger:debug("[ws_channel] auth uuid: ~p", [UUID]), + {noreply, State#state{uuid = UUID, is_authed = true, host_pid = HostPid}} + else {error, Reason} -> + logger:warning("[ws_channel] uuid: ~p, auth failed with reason: ~p", [UUID, Reason]), + {stop, Reason, State} end; handle_request_frame(Ref, {container, ContainerCommand}, State) -> logger:warning("[ws_channel] unsupported request message type: container, ref: ~p, command: ~p", [Ref, ContainerCommand]), @@ -257,16 +234,6 @@ send_reply_frame(Transport, Socket, Ref, Reply) -> Packet = term_to_binary({response, Ref, Reply}), Transport:send(Socket, Packet). --spec stop_after_reply(module(), any(), reference(), tuple(), term(), #state{}) -> {stop, term(), #state{}}. -stop_after_reply(Transport, Socket, Ref, Reply, StopReason, State) -> - case send_reply_frame(Transport, Socket, Ref, Reply) of - ok -> - {stop, StopReason, State}; - {error, Reason} -> - logger:warning("[ws_channel] send response before stop failed, ref: ~p, reason: ~p", [Ref, Reason]), - {stop, {send_failed, Reason}, State} - end. - -spec decode_command_response({container, ok | {ok, term()} | {error, term()}} | tuple()) -> ok | {ok, term()} | {error, term()}. decode_command_response({container, ok}) -> @@ -327,9 +294,3 @@ auth(Token, UUID, Timestamp) when is_binary(Token), is_binary(UUID), is_integer( false -> {error, <<"invalid timestamp">>} end. - --spec reason_to_binary(term()) -> binary(). -reason_to_binary(Reason) when is_binary(Reason) -> - Reason; -reason_to_binary(Reason) -> - unicode:characters_to_binary(io_lib:format("~p", [Reason])).