fix ssl protol
This commit is contained in:
parent
f0facb3691
commit
bc0f906f66
@ -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()}.
|
||||
|
||||
@ -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', _} ->
|
||||
|
||||
@ -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()) ->
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user