diff --git a/apps/iot/src/websocket/tcp_channel.erl b/apps/iot/src/websocket/tcp_channel.erl index 7eb770e..03e9577 100644 --- a/apps/iot/src/websocket/tcp_channel.erl +++ b/apps/iot/src/websocket/tcp_channel.erl @@ -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, <>), + {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, <>), {noreply, State}. %% auth验证 @@ -118,44 +126,48 @@ handle_info({tcp, Socket, <>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) -> +handle_info({tcp, Socket, <>}, 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, <>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) -> - iot_host:handle(HostPid, {ping, CipherMetric}), - {ok, State}; +handle_info({tcp, Socket, <>}, 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, <>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) -> - iot_host:handle(HostPid, {inform, CipherInfo}), - {ok, State}; +handle_info({tcp, Socket, <>}, 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, <>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) -> - iot_host:handle(HostPid, {feedback_step, CipherInfo}), - {ok, State}; +handle_info({tcp, Socket, <>}, 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, <>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) -> - iot_host:handle(HostPid, {feedback_result, CipherInfo}), - {ok, State}; +handle_info({tcp, Socket, <>}, 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, <>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) -> - iot_host:handle(HostPid, {event, CipherEvent}), - {ok, State}; +handle_info({tcp, Socket, <>}, 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, <>}, State = #state{host_pid = HostPid}) when is_pid(HostPid) -> - iot_host:handle(HostPid, {ai_event, CipherEvent}), - {ok, State}; +handle_info({tcp, Socket, <>}, 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, <>}, State = #state{uuid = UUID}) -> - lager:debug("[ws_channel] uuid: ~p, get send response message: ~p", [UUID, Body]), - {ok, State}; -handle_info({binary, <>}, State = #state{uuid = UUID, inflight = Inflight}) when PacketId > 0 -> +handle_info({tcp, Socket, <>}, 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, <>}, 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}. @@ -200,28 +206,4 @@ terminate(Reason, #state{}) -> ok. 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, <>}, State#state{packet_id = PacketId + 1, inflight = NInflight}}; - -%% 发送消息, 不需要等待回复 -websocket_info({send, Msg}, State) when is_binary(Msg) -> - {reply, {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}. \ No newline at end of file