fix tcp_channel
This commit is contained in:
parent
df8cc6f108
commit
aa9b4dde7d
@ -36,20 +36,20 @@
|
||||
-spec publish(Pid :: pid(), ReceiverPid :: pid(), Msg :: binary()) -> Ref :: reference().
|
||||
publish(Pid, ReceiverPid, Msg) when is_pid(Pid), is_binary(Msg) ->
|
||||
Ref = make_ref(),
|
||||
Pid ! {publish, ReceiverPid, Ref, Msg},
|
||||
gen_server:cast(Pid, {publish, ReceiverPid, Ref, Msg}),
|
||||
Ref.
|
||||
|
||||
%% 向通道中写入消息
|
||||
-spec send(Pid :: pid(), Msg :: binary()) -> no_return().
|
||||
send(Pid, Msg) when is_pid(Pid), is_binary(Msg) ->
|
||||
Pid ! {send, Msg}.
|
||||
gen_server:cast(Pid, {send, Msg}).
|
||||
|
||||
%% 关闭方法
|
||||
-spec stop(Pid :: pid(), Reason :: any()) -> no_return().
|
||||
stop(undefined, _Reason) ->
|
||||
ok;
|
||||
stop(Pid, Reason) when is_pid(Pid) ->
|
||||
Pid ! {stop, Reason}.
|
||||
gen_server:stop(Pid, Reason, 5000).
|
||||
|
||||
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
|
||||
%% 逻辑处理方法
|
||||
@ -72,7 +72,15 @@ init([Transport, Sock]) ->
|
||||
handle_call(_Request, _From, State) ->
|
||||
{reply, ok, State}.
|
||||
|
||||
handle_cast(_Msg, State) ->
|
||||
%% 发送消息
|
||||
handle_cast({publish, ReceiverPid, Ref, Msg}, State = #state{transport = Transport, socket = Socket, packet_id = PacketId, inflight = Inflight}) when is_binary(Msg) ->
|
||||
NInflight = maps:put(PacketId, {ReceiverPid, Ref}, Inflight),
|
||||
Transport:send(Socket, <<?PACKET_PUBLISH, PacketId:32, Msg/binary>>),
|
||||
{noreply, State#state{packet_id = PacketId + 1, inflight = NInflight}};
|
||||
|
||||
%% 发送消息, 不需要等待回复
|
||||
handle_cast({send, Msg}, State = #state{transport = Transport, socket = Socket}) when is_binary(Msg) ->
|
||||
Transport:send(Socket, <<?PACKET_PUBLISH, 0:32, Msg/binary>>),
|
||||
{noreply, State}.
|
||||
|
||||
%% auth验证
|
||||
@ -118,44 +126,48 @@ handle_info({tcp, Socket, <<?PACKET_REQUEST, PacketId:32, ?METHOD_AUTH:8, AuthRe
|
||||
{stop, State}
|
||||
end;
|
||||
|
||||
handle_info({binary, <<?PACKET_REQUEST, _PacketId:32, ?METHOD_DATA:8, Data/binary>>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
handle_info({tcp, Socket, <<?PACKET_REQUEST, 0:32, ?METHOD_DATA:8, Data0/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
Data = message_pb:decode_msg(Data0, data),
|
||||
iot_host:handle(HostPid, {data, Data}),
|
||||
{ok, State};
|
||||
{noreply, State};
|
||||
|
||||
handle_info({binary, <<?PACKET_REQUEST, _PacketId:32, ?METHOD_PING:8, CipherMetric/binary>>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
iot_host:handle(HostPid, {ping, CipherMetric}),
|
||||
{ok, State};
|
||||
handle_info({tcp, Socket, <<?PACKET_REQUEST, 0:32, ?METHOD_PING:8, PingData/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
Ping = message_pb:decode_msg(PingData, ping),
|
||||
iot_host:handle(HostPid, {ping, Ping}),
|
||||
{noreply, State};
|
||||
|
||||
handle_info({binary, <<?PACKET_REQUEST, _PacketId:32, ?METHOD_INFORM:8, CipherInfo/binary>>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
iot_host:handle(HostPid, {inform, CipherInfo}),
|
||||
{ok, State};
|
||||
handle_info({tcp, Socket, <<?PACKET_REQUEST, 0:32, ?METHOD_INFORM:8, InformData/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
ServiceInform = message_pb:decode_msg(InformData, service_inform),
|
||||
iot_host:handle(HostPid, {inform, ServiceInform}),
|
||||
{noreply, State};
|
||||
|
||||
handle_info({binary, <<?PACKET_REQUEST, _PacketId:32, ?METHOD_FEEDBACK_STEP:8, CipherInfo/binary>>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
iot_host:handle(HostPid, {feedback_step, CipherInfo}),
|
||||
{ok, State};
|
||||
handle_info({tcp, Socket, <<?PACKET_REQUEST, 0:32, ?METHOD_FEEDBACK_STEP:8, FeedbackStepData/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
FeedbackStep = message_pb:decode_msg(FeedbackStepData, feedback_step),
|
||||
iot_host:handle(HostPid, {feedback_step, FeedbackStep}),
|
||||
{noreply, State};
|
||||
|
||||
handle_info({binary, <<?PACKET_REQUEST, _PacketId:32, ?METHOD_FEEDBACK_RESULT:8, CipherInfo/binary>>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
iot_host:handle(HostPid, {feedback_result, CipherInfo}),
|
||||
{ok, State};
|
||||
handle_info({tcp, Socket, <<?PACKET_REQUEST, 0:32, ?METHOD_FEEDBACK_RESULT:8, FeedbackResultData/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
FeedbackResult = message_pb:decode_msg(FeedbackResultData, feedback_result),
|
||||
iot_host:handle(HostPid, {feedback_result, FeedbackResult}),
|
||||
{noreply, State};
|
||||
|
||||
handle_info({binary, <<?PACKET_REQUEST, _PacketId:32, ?METHOD_EVENT:8, CipherEvent/binary>>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
iot_host:handle(HostPid, {event, CipherEvent}),
|
||||
{ok, State};
|
||||
handle_info({tcp, Socket, <<?PACKET_REQUEST, 0:32, ?METHOD_EVENT:8, EventData/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
Event = message_pb:decode_msg(EventData, event),
|
||||
iot_host:handle(HostPid, {event, Event}),
|
||||
{noreply, State};
|
||||
|
||||
handle_info({binary, <<?PACKET_REQUEST, _PacketId:32, ?METHOD_AI_EVENT:8, CipherEvent/binary>>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
iot_host:handle(HostPid, {ai_event, CipherEvent}),
|
||||
{ok, State};
|
||||
handle_info({tcp, Socket, <<?PACKET_REQUEST, 0:32, ?METHOD_AI_EVENT:8, AIEventData/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
AIEvent = message_pb:decode_msg(AIEventData, ai_event),
|
||||
iot_host:handle(HostPid, {ai_event, AIEvent}),
|
||||
{noreply, State};
|
||||
|
||||
%% 主机端的消息响应
|
||||
handle_info({binary, <<?PACKET_PUBLISH_RESPONSE, 0:32, Body/binary>>}, State = #state{uuid = UUID}) ->
|
||||
lager:debug("[ws_channel] uuid: ~p, get send response message: ~p", [UUID, Body]),
|
||||
{ok, State};
|
||||
handle_info({binary, <<?PACKET_PUBLISH_RESPONSE, PacketId:32, Body/binary>>}, State = #state{uuid = UUID, inflight = Inflight}) when PacketId > 0 ->
|
||||
handle_info({tcp, Socket, <<?PACKET_PUBLISH_RESPONSE, PacketId:32, Body/binary>>}, State = #state{socket = Socket, uuid = UUID, inflight = Inflight}) when PacketId > 0 ->
|
||||
lager:debug("[ws_channel] uuid: ~p, get publish response message: ~p, packet_id: ~p", [UUID, Body, PacketId]),
|
||||
case maps:take(PacketId, Inflight) of
|
||||
error ->
|
||||
lager:warning("[ws_channel] get unknown publish response message: ~p, packet_id: ~p", [Body, PacketId]),
|
||||
{ok, State};
|
||||
{noreply, State};
|
||||
{{ReceiverPid, Ref}, NInflight} ->
|
||||
case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of
|
||||
true when Body == <<>> ->
|
||||
@ -165,20 +177,9 @@ handle_info({binary, <<?PACKET_PUBLISH_RESPONSE, PacketId:32, Body/binary>>}, St
|
||||
false ->
|
||||
lager:warning("[ws_channel] get publish response message: ~p, packet_id: ~p, but receiver_pid is deaded", [Body, PacketId])
|
||||
end,
|
||||
{ok, State#state{inflight = NInflight}}
|
||||
{noreply, State#state{inflight = NInflight}}
|
||||
end;
|
||||
|
||||
handle_info({tcp, Socket, Data}, State = #state{transport = Transport, socket = Socket}) ->
|
||||
lager:debug("[ws_channel] socket: ~p, get a message: ~p", [Socket, Data]),
|
||||
|
||||
Transport:send(Socket, <<"hello world reply">>),
|
||||
|
||||
{noreply, State};
|
||||
|
||||
handle_info(Info, State) ->
|
||||
lager:error("[ws_channel] get a unknown message: ~p, channel will closed", [Info]),
|
||||
{stop, State};
|
||||
|
||||
handle_info({tcp_error, Sock, Reason}, State = #state{socket = Sock}) ->
|
||||
lager:notice("[sdlan_channel] tcp_error: ~p", [Reason]),
|
||||
{stop, normal, State};
|
||||
@ -191,6 +192,11 @@ handle_info({tcp_closed, Sock}, State = #state{socket = Sock}) ->
|
||||
handle_info({stop, Reason}, State) ->
|
||||
{stop, Reason, State};
|
||||
|
||||
%% 主机进程挂掉的时候，需要关闭掉链接
|
||||
handle_info({'DOWN', _, process, HostPid, Reason}, State = #state{uuid = UUID, host_pid = HostPid}) ->
|
||||
lager:debug("[ws_channel] uuid: ~p, channel will close because host exited with reason: ~p", [UUID, Reason]),
|
||||
{stop, State};
|
||||
|
||||
handle_info(Info, State) ->
|
||||
lager:warning("[sdlan_channel] get a unknown message: ~p, channel will closed", [Info]),
|
||||
{noreply, State}.
|
||||
@ -201,27 +207,3 @@ terminate(Reason, #state{}) ->
|
||||
|
||||
code_change(_OldVsn, State, _Extra) ->
|
||||
{ok, State}.
|
||||
|
||||
%% 处理关闭信号
|
||||
websocket_info({stop, Reason}, State) ->
|
||||
lager:debug("[ws_channel] the channel will be closed with reason: ~p", [Reason]),
|
||||
{stop, State};
|
||||
|
||||
%% 发送消息
|
||||
websocket_info({publish, ReceiverPid, Ref, Msg}, State = #state{packet_id = PacketId, inflight = Inflight}) when is_binary(Msg) ->
|
||||
NInflight = maps:put(PacketId, {ReceiverPid, Ref}, Inflight),
|
||||
{reply, {binary, <<?PACKET_PUBLISH, PacketId:32, Msg/binary>>}, State#state{packet_id = PacketId + 1, inflight = NInflight}};
|
||||
|
||||
%% 发送消息, 不需要等待回复
|
||||
websocket_info({send, Msg}, State) when is_binary(Msg) ->
|
||||
{reply, {binary, <<?PACKET_PUBLISH, 0:32, Msg/binary>>}, State};
|
||||
|
||||
%% 用户进程关闭，则关闭通道
|
||||
websocket_info({'DOWN', _, process, HostPid, Reason}, State = #state{uuid = UUID, host_pid = HostPid}) ->
|
||||
lager:debug("[ws_channel] uuid: ~p, channel will close because host exited with reason: ~p", [UUID, Reason]),
|
||||
{stop, State};
|
||||
|
||||
%% 处理其他未知消息
|
||||
websocket_info(Info, State = #state{uuid = UUID}) ->
|
||||
lager:debug("[ws_channel] channel get unknown info: ~p, uuid: ~p", [Info, UUID]),
|
||||
{ok, State}.
|
||||
Loading…
x
Reference in New Issue
Block a user