diff --git a/src/host/iot_host.erl b/src/host/iot_host.erl index 29ae5f1..b55a63a 100644 --- a/src/host/iot_host.erl +++ b/src/host/iot_host.erl @@ -125,17 +125,17 @@ remove_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) container_call(Pid, docker_container_builder:remove_request(ContainerName)). -spec await_reply(Pid :: pid(), Ref :: reference(), Timeout :: integer()) -> - {ok, Result :: term()} | {error, Code :: integer(), Reason :: binary()}. + {ok, Result :: term()} | {error, Reason :: term()}. await_reply(Pid, Ref, Timeout) when is_pid(Pid), is_reference(Ref), is_integer(Timeout) -> receive {request_reply, Ref, {ok, Result}} -> {ok, Result}; - {request_reply, Ref, {error, Code, Reason}} when is_integer(Code), is_binary(Reason) -> - {error, Code, Reason} + {request_reply, Ref, {error, Reason}} -> + {error, Reason} after Timeout -> ok = gen_statem:call(Pid, {cancel_request_call, Ref}), flush_reply(Ref), - {error, -1, <<"timeout">>} + {error, timeout} end. -spec pub(Pid :: pid(), Topic :: binary(), Qos :: integer(), Content :: binary()) -> ok | {error, Reason :: any()}. diff --git a/src/transport/http/container_handler.erl b/src/transport/http/container_handler.erl index c982f9e..8d7e319 100644 --- a/src/transport/http/container_handler.erl +++ b/src/transport/http/container_handler.erl @@ -26,8 +26,8 @@ handle_request("GET", "/container/get_all", #{<<"uuid">> := UUID}, _) when is_bi case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of {ok, Result} -> {ok, 200, request_success_response(Result)}; - {error, Code, Reason} -> - request_error_http_response(Code, Reason) + {error, Reason} -> + request_error_http_response(Reason) end; {error, Reason} when is_binary(Reason) -> {ok, 200, iot_util:json_error(-1, Reason)} @@ -51,8 +51,8 @@ handle_request("POST", "/container/push_config", _, case iot_host:await_reply(Pid, Ref, Timeout) of {ok, Result} -> {ok, 200, request_success_response(Result)}; - {error, Code, Reason} -> - request_error_http_response(Code, Reason) + {error, Reason} -> + request_error_http_response(Reason) end; {error, Reason} when is_binary(Reason) -> {ok, 200, iot_util:json_error(-1, Reason)} @@ -71,8 +71,8 @@ handle_request("POST", "/container/deploy", _, #{<<"uuid">> := UUID, <<"task_id" case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of {ok, Result} -> {ok, 200, request_success_response(Result)}; - {error, Code, Reason} -> - request_error_http_response(Code, Reason) + {error, Reason} -> + request_error_http_response(Reason) end; {error, Reason} when is_binary(Reason) -> {ok, 200, iot_util:json_error(400, Reason)} @@ -90,8 +90,8 @@ handle_request("POST", "/container/start", _, #{<<"uuid">> := UUID, <<"container case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of {ok, Result} -> {ok, 200, request_success_response(Result)}; - {error, Code, Reason} -> - request_error_http_response(Code, Reason) + {error, Reason} -> + request_error_http_response(Reason) end; {error, Reason} when is_binary(Reason) -> {ok, 200, iot_util:json_error(400, Reason)} @@ -109,8 +109,8 @@ handle_request("POST", "/container/stop", _, #{<<"uuid">> := UUID, <<"container_ case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of {ok, Result} -> {ok, 200, request_success_response(Result)}; - {error, Code, Reason} -> - request_error_http_response(Code, Reason) + {error, Reason} -> + request_error_http_response(Reason) end; {error, Reason} when is_binary(Reason) -> {ok, 200, iot_util:json_error(400, Reason)} @@ -127,8 +127,8 @@ handle_request("POST", "/container/kill", _, #{<<"uuid">> := UUID, <<"container_ case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of {ok, Result} -> {ok, 200, request_success_response(Result)}; - {error, Code, Reason} -> - request_error_http_response(Code, Reason) + {error, Reason} -> + request_error_http_response(Reason) end; {error, Reason} when is_binary(Reason) -> {ok, 200, iot_util:json_error(400, Reason)} @@ -146,8 +146,8 @@ handle_request("POST", "/container/remove", _, #{<<"uuid">> := UUID, <<"containe case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of {ok, Result} -> {ok, 200, request_success_response(Result)}; - {error, Code, Reason} -> - request_error_http_response(Code, Reason) + {error, Reason} -> + request_error_http_response(Reason) end; {error, Reason} when is_binary(Reason) -> {ok, 200, iot_util:json_error(400, Reason)} @@ -178,19 +178,30 @@ request_error_response(Code, Reason) when is_integer(Code), is_binary(Reason) -> iot_util:json_error(Code, Reason) end. --spec request_error_http_response(Code :: integer(), Reason :: binary()) -> +-spec request_error_http_response(Reason :: term()) -> {ok, HttpStatus :: integer(), Body :: iolist()}. -request_error_http_response(Code, Reason) when is_integer(Code), is_binary(Reason) -> - {ok, request_error_status(Code), request_error_response(Code, Reason)}. +request_error_http_response(Reason) -> + HttpStatus = request_error_status(Reason), + {ok, HttpStatus, request_error_response(HttpStatus, reason_to_binary(Reason))}. --spec request_error_status(Code :: integer()) -> integer(). -request_error_status(Code) when is_integer(Code), Code >= 400, Code < 600 -> - Code; -request_error_status(-1) -> +-spec request_error_status(Reason :: term()) -> integer(). +request_error_status(timeout) -> + 504; +request_error_status(<<"timeout">>) -> 504; request_error_status(_) -> 400. +-spec reason_to_binary(term()) -> binary(). +reason_to_binary(Reason) when is_binary(Reason) -> + Reason; +reason_to_binary(timeout) -> + <<"timeout">>; +reason_to_binary(invalid_response) -> + <<"invalid response">>; +reason_to_binary(Reason) -> + unicode:characters_to_binary(io_lib:format("~p", [Reason])). + decode_json_bytes(Data) when is_binary(Data) -> case catch json:decode(Data) of {'EXIT', _} -> diff --git a/src/transport/tcp/ssl_channel.erl b/src/transport/tcp/ssl_channel.erl index ba2e9b3..913fdfa 100644 --- a/src/transport/tcp/ssl_channel.erl +++ b/src/transport/tcp/ssl_channel.erl @@ -10,7 +10,6 @@ -author("licheng5"). -behaviour(ranch_protocol). --define(MAX_PACKET_ID, 16#FFFFFFFF). -define(INFLIGHT_TIMEOUT, 60000). %% API @@ -27,8 +26,6 @@ uuid :: undefined | binary(), %% 用户进程id host_pid = undefined, - %% 发送消息对应的id - packet_id = 1 :: integer(), %% 请求响应的对应关系 inflight = #{} @@ -36,7 +33,6 @@ -record(inflight_request, { receiver_pid :: pid(), - ref :: reference(), timer_ref :: reference() }). @@ -88,8 +84,8 @@ init(Ref, Transport, _Opts = []) -> gen_server:enter_loop(?MODULE, [], #state{transport = Transport, socket = Socket}). handle_call({cancel_request_call, Ref}, _From, State = #state{inflight = Inflight}) -> - case take_inflight_by_ref(Ref, Inflight) of - {ok, #inflight_request{timer_ref = TimerRef}, NInflight} -> + case maps:take(Ref, Inflight) of + {#inflight_request{timer_ref = TimerRef}, NInflight} -> erlang:cancel_timer(TimerRef), {reply, ok, State#state{inflight = NInflight}}; error -> @@ -111,22 +107,19 @@ handle_cast({command, Command}, State = #state{transport = Transport, socket = S {noreply, State}; %% 推送需要响应的请求 -handle_cast({request_call, ReceiverPid, Ref, Body}, State = #state{transport = Transport, socket = Socket, packet_id = PacketId, inflight = Inflight}) -> - Packet = term_to_binary({request, PacketId, Body}), - TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {request_timeout, PacketId}), +handle_cast({request_call, ReceiverPid, Ref, Body}, State = #state{transport = Transport, socket = Socket, inflight = Inflight}) -> + Packet = term_to_binary({request, Ref, Body}), + TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {request_timeout, Ref}), Transport:send(Socket, Packet), - RequestInfo = #inflight_request{receiver_pid = ReceiverPid, ref = Ref, timer_ref = TimerRef}, - {noreply, State#state{ - packet_id = inc_packet_id(PacketId), - inflight = maps:put(PacketId, RequestInfo, Inflight) - }}. + RequestInfo = #inflight_request{receiver_pid = ReceiverPid, timer_ref = TimerRef}, + {noreply, State#state{inflight = maps:put(Ref, RequestInfo, Inflight)}}. -handle_info({timeout, TimerRef, {request_timeout, PacketId}}, State = #state{inflight = Inflight}) -> - case maps:get(PacketId, Inflight, undefined) of - #inflight_request{ref = Ref, timer_ref = TimerRef} -> - logger:warning("[ws_channel] request timeout, packet_id: ~p, ref: ~p", [PacketId, Ref]), - {noreply, State#state{inflight = maps:remove(PacketId, Inflight)}}; +handle_info({timeout, TimerRef, {request_timeout, Ref}}, State = #state{inflight = Inflight}) -> + case maps:get(Ref, Inflight, undefined) of + #inflight_request{timer_ref = TimerRef} -> + logger:warning("[ws_channel] request timeout, ref: ~p", [Ref]), + {noreply, State#state{inflight = maps:remove(Ref, Inflight)}}; _ -> {noreply, State} end; @@ -143,12 +136,12 @@ handle_info({'DOWN', _, process, HostPid, Reason}, State = #state{uuid = UUID, h handle_info({ssl, Socket, PacketBin}, State = #state{transport = Transport, socket = Socket, host_pid = HostPid, inflight = Inflight}) when is_binary(PacketBin) -> case catch binary_to_term(PacketBin, [safe]) of - {request, PacketId, Body} -> - handle_request_frame(PacketId, Body, Transport, Socket, State); + {request, Ref, Body} -> + handle_request_frame(Ref, Body, Transport, Socket, State); {message, Body} -> handle_message_frame(Body, HostPid, State); - {response, PacketId, Response} -> - handle_response_frame(PacketId, Response, Inflight, State); + {response, Ref, Response} -> + handle_response_frame(Ref, Response, Inflight, State); {'EXIT', Reason} -> logger:warning("[ssl_channel] invalid packet: ~p", [Reason]), {stop, bad_packet, State}; @@ -185,26 +178,8 @@ code_change(_OldVsn, State, _Extra) -> %%%% helper methods %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% --spec take_inflight_by_ref(reference(), map()) -> - error | {ok, #inflight_request{}, map()}. -take_inflight_by_ref(Ref, Inflight) -> - maps:fold(fun(PacketId, Request = #inflight_request{ref = PacketRef}, Acc) -> - case Acc of - error when PacketRef =:= Ref -> - {ok, Request, maps:remove(PacketId, Inflight)}; - _ -> - Acc - end - end, error, Inflight). - --spec inc_packet_id(integer()) -> integer(). -inc_packet_id(?MAX_PACKET_ID) -> - 1; -inc_packet_id(PacketId) when PacketId > 0, PacketId < ?MAX_PACKET_ID -> - PacketId + 1. - --spec handle_request_frame(non_neg_integer(), tuple(), module(), any(), #state{}) -> {noreply, #state{}} | {stop, term(), #state{}}. -handle_request_frame(PacketId, +-spec handle_request_frame(reference(), tuple(), module(), any(), #state{}) -> {noreply, #state{}} | {stop, term(), #state{}}. +handle_request_frame(Ref, {auth_request, #{uuid := UUID, token := Token, timestamp := Timestamp}}, Transport, Socket, State) -> @@ -221,29 +196,29 @@ handle_request_frame(PacketId, case iot_host:attach_channel(HostPid, self()) of ok -> erlang:monitor(process, HostPid), - send_reply_frame(Transport, Socket, PacketId, {auth_response, {ok, <<"ok">>}}), + send_reply_frame(Transport, Socket, Ref, {auth_response, {ok, <<"ok">>}}), {noreply, State#state{uuid = UUID, host_pid = HostPid}}; {denied, Reason} when is_binary(Reason) -> erlang:monitor(process, HostPid), - send_reply_frame(Transport, Socket, PacketId, {auth_response, {error, 1, Reason}}), + send_reply_frame(Transport, Socket, Ref, {auth_response, {error, {denied, Reason}}}), logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]), {noreply, State#state{uuid = UUID, host_pid = HostPid}}; {error, Reason} when is_binary(Reason) -> - send_reply_frame(Transport, Socket, PacketId, {auth_response, {error, 2, Reason}}), + send_reply_frame(Transport, Socket, Ref, {auth_response, {error, {failed, Reason}}}), logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]), {stop, Reason, State} end end; {error, Reason} -> - send_reply_frame(Transport, Socket, PacketId, {auth_response, {error, 2, Reason}}), + send_reply_frame(Transport, Socket, Ref, {auth_response, {error, {failed, Reason}}}), logger:warning("[ws_channel] uuid: ~p, token: ~p, auth failed, reason: ~p", [UUID, Token, Reason]), {stop, Reason, State} end; -handle_request_frame(PacketId, {container_request, ContainerRequest}, _Transport, _Socket, State) -> - logger:warning("[ws_channel] unsupported request message type: container_request, packet_id: ~p, request: ~p", [PacketId, ContainerRequest]), +handle_request_frame(Ref, {container_request, ContainerRequest}, _Transport, _Socket, State) -> + logger:warning("[ws_channel] unsupported request message type: container_request, ref: ~p, request: ~p", [Ref, ContainerRequest]), {stop, normal, State}; -handle_request_frame(PacketId, Body, _Transport, _Socket, State) -> - logger:warning("[ws_channel] unsupported request body, packet_id: ~p, body: ~p", [PacketId, Body]), +handle_request_frame(Ref, Body, _Transport, _Socket, State) -> + logger:warning("[ws_channel] unsupported request body, ref: ~p, body: ~p", [Ref, Body]), {stop, normal, State}. -spec handle_message_frame(tuple(), undefined | pid(), #state{}) -> @@ -265,41 +240,39 @@ handle_event_stream_frame(#{task_id := TaskId, type := Type, stream := Stream}) logger:debug("[ssl_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, Type, Stream]), iot_event_stream_observer:stream_data(TaskId, Type, Stream). --spec handle_response_frame(non_neg_integer(), tuple(), map(), #state{}) -> +-spec handle_response_frame(reference(), tuple(), map(), #state{}) -> {noreply, #state{}}. -handle_response_frame(PacketId, Reply, Inflight, State) when PacketId > 0 -> - case maps:take(PacketId, Inflight) of +handle_response_frame(Ref, Reply, Inflight, State) when is_reference(Ref) -> + case maps:take(Ref, Inflight) of error -> {noreply, State}; - {#inflight_request{receiver_pid = ReceiverPid, ref = Ref, timer_ref = TimerRef}, NInflight} -> + {#inflight_request{receiver_pid = ReceiverPid, timer_ref = TimerRef}, NInflight} -> erlang:cancel_timer(TimerRef), case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of true -> ReceiverPid ! {request_reply, Ref, decode_reply(Reply)}; false -> - logger:warning("[ws_channel] get reply message: ~p, packet_id: ~p, but receiver_pid is deaded", [Reply, PacketId]) + logger:warning("[ws_channel] get reply message: ~p, ref: ~p, but receiver_pid is deaded", [Reply, Ref]) end, {noreply, State#state{inflight = NInflight}} end; -handle_response_frame(PacketId, Reply, _Inflight, State) -> - logger:warning("[ws_channel] unexpected response frame, packet_id: ~p, reply: ~p", [PacketId, Reply]), +handle_response_frame(Ref, Reply, _Inflight, State) -> + logger:warning("[ws_channel] unexpected response frame, ref: ~p, reply: ~p", [Ref, Reply]), {noreply, State}. --spec send_reply_frame(module(), any(), non_neg_integer(), tuple()) -> any(). -send_reply_frame(Transport, Socket, PacketId, Reply) -> - Packet = term_to_binary({response, PacketId, Reply}), +-spec send_reply_frame(module(), any(), reference(), tuple()) -> any(). +send_reply_frame(Transport, Socket, Ref, Reply) -> + Packet = term_to_binary({response, Ref, Reply}), Transport:send(Socket, Packet). --spec decode_reply({container_response, {ok, term()} | {error, integer(), binary()}} | tuple()) -> - {ok, term()} | {error, integer(), binary()} | undefined. +-spec decode_reply({container_response, {ok, term()} | {error, term()}} | tuple()) -> + {ok, term()} | {error, term()}. decode_reply({container_response, {ok, Result}}) -> {ok, Result}; -decode_reply({container_response, {error, Code, Message}}) -> - {error, Code, Message}; -decode_reply(undefined) -> - undefined; +decode_reply({container_response, {error, Reason}}) -> + {error, Reason}; decode_reply(_Reply) -> - undefined. + {error, invalid_response}. %% 检测token是否是合法值 -spec auth(Token :: binary(), UUID :: binary(), Timestamp :: integer()) ->