fix ssl_channel
This commit is contained in:
parent
e0b2da5e3f
commit
509d679d70
@ -117,14 +117,14 @@ handle_info({'DOWN', _, process, HostPid, Reason}, State = #state{uuid = UUID, h
|
||||
logger:debug("[ws_channel] uuid: ~p, channel will close because host exited with reason: ~p", [UUID, Reason]),
|
||||
{stop, Reason, State};
|
||||
|
||||
handle_info({ssl, Socket, PacketBin}, State = #state{transport = Transport, socket = Socket, host_pid = HostPid, inflight = Inflight}) when is_binary(PacketBin) ->
|
||||
handle_info({ssl, Socket, PacketBin}, State = #state{socket = Socket}) when is_binary(PacketBin) ->
|
||||
try binary_to_term(PacketBin, [safe]) of
|
||||
{request, Ref, Body} ->
|
||||
handle_request_frame(Ref, Body, Transport, Socket, State);
|
||||
handle_request_frame(Ref, Body, State);
|
||||
{message, Body} ->
|
||||
handle_message_frame(Body, HostPid, State);
|
||||
handle_message_frame(Body, State);
|
||||
{command_response, Ref, Response} ->
|
||||
handle_command_response_frame(Ref, Response, Inflight, State);
|
||||
handle_command_response_frame(Ref, Response, State);
|
||||
Other ->
|
||||
logger:warning("[ssl_channel] unsupported packet: ~p", [Other]),
|
||||
{stop, bad_packet, State}
|
||||
@ -161,10 +161,8 @@ code_change(_OldVsn, State, _Extra) ->
|
||||
%%%% helper methods
|
||||
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
|
||||
|
||||
-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) ->
|
||||
|
||||
-spec handle_request_frame(reference(), tuple(), #state{}) -> {noreply, #state{}} | {stop, term(), #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 ->
|
||||
@ -191,22 +189,22 @@ handle_request_frame(Ref, {auth_request, #{uuid := UUID, token := Token, timesta
|
||||
logger:warning("[ws_channel] uuid: ~p, token: ~p, auth failed, reason: ~p", [UUID, Token, Reason]),
|
||||
{stop, Reason, State}
|
||||
end;
|
||||
handle_request_frame(Ref, {container, ContainerCommand}, _Transport, _Socket, State) ->
|
||||
handle_request_frame(Ref, {container, ContainerCommand}, State) ->
|
||||
logger:warning("[ws_channel] unsupported request message type: container, ref: ~p, command: ~p", [Ref, ContainerCommand]),
|
||||
{stop, normal, State};
|
||||
handle_request_frame(Ref, Body, _Transport, _Socket, State) ->
|
||||
handle_request_frame(Ref, Body, 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{}) ->
|
||||
-spec handle_message_frame(tuple(), #state{}) ->
|
||||
{noreply, #state{}}.
|
||||
handle_message_frame({data, #{route_key := RouteKey, metric := Metric}}, HostPid, State) when is_pid(HostPid) ->
|
||||
handle_message_frame({data, #{route_key := RouteKey, metric := Metric}}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
iot_host:handle(HostPid, {data, RouteKey, Metric}),
|
||||
{noreply, State};
|
||||
handle_message_frame({task_event, Event}, HostPid, State) when is_pid(HostPid) ->
|
||||
handle_message_frame({task_event, Event}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
handle_event_stream_frame(Event),
|
||||
{noreply, State};
|
||||
handle_message_frame(Body, _HostPid, State) ->
|
||||
handle_message_frame(Body, State) ->
|
||||
logger:warning("[ssl_channel] unsupported message body: ~p", [Body]),
|
||||
{noreply, State}.
|
||||
|
||||
@ -217,9 +215,9 @@ 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_command_response_frame(reference(), tuple(), map(), #state{}) ->
|
||||
-spec handle_command_response_frame(reference(), tuple(), #state{}) ->
|
||||
{noreply, #state{}}.
|
||||
handle_command_response_frame(Ref, Reply, Inflight, State) when is_reference(Ref) ->
|
||||
handle_command_response_frame(Ref, Reply, State = #state{inflight = Inflight}) when is_reference(Ref) ->
|
||||
case maps:take(Ref, Inflight) of
|
||||
error ->
|
||||
{noreply, State};
|
||||
@ -228,7 +226,7 @@ handle_command_response_frame(Ref, Reply, Inflight, State) when is_reference(Ref
|
||||
deliver_command_response(ReceiverPid, Ref, Reply),
|
||||
{noreply, State#state{inflight = NInflight}}
|
||||
end;
|
||||
handle_command_response_frame(Ref, Reply, _Inflight, State) ->
|
||||
handle_command_response_frame(Ref, Reply, State) ->
|
||||
logger:warning("[ws_channel] unexpected command_response frame, ref: ~p, reply: ~p", [Ref, Reply]),
|
||||
{noreply, State}.
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user