fix
This commit is contained in:
parent
5d6350864a
commit
e303cd29c0
@ -7,12 +7,13 @@
|
||||
%%%-------------------------------------------------------------------
|
||||
-author("anlicheng").
|
||||
|
||||
%% efka主动发起的消息体类型, 消息大类
|
||||
-define(PACKET_REQUEST, 16#01).
|
||||
-define(PACKET_RESPONSE, 16#02).
|
||||
|
||||
%% efka主动发起不需要返回的数据
|
||||
-define(PACKET_CAST, 16#03).
|
||||
%% 一级帧类型
|
||||
%% REQUEST: 需要响应的请求帧
|
||||
%% RESPONSE: 对 REQUEST 的响应帧
|
||||
%% CAST: 单向消息帧，不需要响应
|
||||
-define(FRAME_REQUEST, 16#01).
|
||||
-define(FRAME_RESPONSE, 16#02).
|
||||
-define(FRAME_CAST, 16#03).
|
||||
|
||||
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
|
||||
%%%% 二级分类定义
|
||||
|
||||
@ -99,13 +99,13 @@ handle_call(_Request, _From, State) ->
|
||||
%% 发送消息, 基于pub/sub机制
|
||||
handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, socket = Socket}) ->
|
||||
Encoded = message_pb:encode_msg(#'Pub'{topic = Topic, qos = Qos, content = Content}),
|
||||
Transport:send(Socket, <<?PACKET_CAST, ?MESSAGE_PUB, Encoded/binary>>),
|
||||
Transport:send(Socket, <<?FRAME_CAST, ?MESSAGE_PUB, Encoded/binary>>),
|
||||
{noreply, State};
|
||||
|
||||
%% 发送Command消息
|
||||
handle_cast({command, CommandType, Command}, State = #state{transport = Transport, socket = Socket}) ->
|
||||
Encoded = message_pb:encode_msg(#'Command'{command_type = CommandType, command = Command}),
|
||||
Transport:send(Socket, <<?PACKET_CAST, ?MESSAGE_COMMAND, Encoded/binary>>),
|
||||
Transport:send(Socket, <<?FRAME_CAST, ?MESSAGE_COMMAND, Encoded/binary>>),
|
||||
{noreply, State};
|
||||
|
||||
%% 推送消息
|
||||
@ -119,7 +119,7 @@ handle_cast({jsonrpc_call, ReceiverPid, Ref, {Method, Params}}, State = #state{t
|
||||
|
||||
EncRequest = <<?MESSAGE_JSONRPC_REQUEST, Encoded/binary>>,
|
||||
TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {jsonrpc_timeout, NPacketId}),
|
||||
Transport:send(Socket, <<?PACKET_REQUEST, NPacketId:32, EncRequest/binary>>),
|
||||
Transport:send(Socket, <<?FRAME_REQUEST, NPacketId:32, EncRequest/binary>>),
|
||||
|
||||
RequestInfo = #inflight_request{receiver_pid = ReceiverPid, ref = Ref, timer_ref = TimerRef},
|
||||
{noreply, State#state{packet_id = NextPacketId, inflight = maps:put(NPacketId, RequestInfo, Inflight)}};
|
||||
@ -129,7 +129,7 @@ handle_cast({jsonrpc_call, ReceiverPid, Ref, {Method, Params}}, State = #state{t
|
||||
end.
|
||||
|
||||
%% auth验证
|
||||
handle_info({tcp, Socket, <<?PACKET_REQUEST, PacketId:32, ?MESSAGE_AUTH_REQUEST, RequestBin/binary>>}, State = #state{transport = Transport, socket = Socket}) ->
|
||||
handle_info({tcp, Socket, <<?FRAME_REQUEST, PacketId:32, ?MESSAGE_AUTH_REQUEST, RequestBin/binary>>}, State = #state{transport = Transport, socket = Socket}) ->
|
||||
#'AuthRequest'{uuid = UUID, username = Username, token = Token, salt = Salt, timestamp = Timestamp} = message_pb:decode_msg(RequestBin, 'AuthRequest'),
|
||||
|
||||
logger:debug("[ws_channel] auth uuid: ~p", [UUID]),
|
||||
@ -148,14 +148,14 @@ handle_info({tcp, Socket, <<?PACKET_REQUEST, PacketId:32, ?MESSAGE_AUTH_REQUEST,
|
||||
erlang:monitor(process, HostPid),
|
||||
Encoded = message_pb:encode_msg(#'AuthReply'{code = 0, payload = <<"ok">>}, 'AuthReply'),
|
||||
AuthReplyBin = <<?MESSAGE_AUTH_REPLY, Encoded/binary>>,
|
||||
Transport:send(Socket, <<?PACKET_RESPONSE, PacketId:32, AuthReplyBin/binary>>),
|
||||
Transport:send(Socket, <<?FRAME_RESPONSE, PacketId:32, AuthReplyBin/binary>>),
|
||||
|
||||
{noreply, State#state{uuid = UUID, host_pid = HostPid}};
|
||||
{denied, Reason} when is_binary(Reason) ->
|
||||
erlang:monitor(process, HostPid),
|
||||
Encoded = message_pb:encode_msg(#'AuthReply'{code = 1, payload = Reason}, 'AuthReply'),
|
||||
AuthReplyBin = <<?MESSAGE_AUTH_REPLY, Encoded/binary>>,
|
||||
Transport:send(Socket, <<?PACKET_RESPONSE, PacketId:32, AuthReplyBin/binary>>),
|
||||
Transport:send(Socket, <<?FRAME_RESPONSE, PacketId:32, AuthReplyBin/binary>>),
|
||||
logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]),
|
||||
|
||||
{noreply, State#state{uuid = UUID, host_pid = HostPid}};
|
||||
@ -163,7 +163,7 @@ handle_info({tcp, Socket, <<?PACKET_REQUEST, PacketId:32, ?MESSAGE_AUTH_REQUEST,
|
||||
{error, Reason} when is_binary(Reason) ->
|
||||
Encoded = message_pb:encode_msg(#'AuthReply'{code = 2, payload = Reason}, 'AuthReply'),
|
||||
AuthReplyBin = <<?MESSAGE_AUTH_REPLY, Encoded/binary>>,
|
||||
Transport:send(Socket, <<?PACKET_RESPONSE, PacketId:32, AuthReplyBin/binary>>),
|
||||
Transport:send(Socket, <<?FRAME_RESPONSE, PacketId:32, AuthReplyBin/binary>>),
|
||||
logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]),
|
||||
|
||||
{stop, State}
|
||||
@ -173,16 +173,16 @@ handle_info({tcp, Socket, <<?PACKET_REQUEST, PacketId:32, ?MESSAGE_AUTH_REQUEST,
|
||||
logger:warning("[ws_channel] uuid: ~p, user: ~p, auth failed", [UUID, Username]),
|
||||
{stop, State}
|
||||
end;
|
||||
handle_info({tcp, Socket, <<?PACKET_REQUEST, _PacketId:32, MsgType:8, _/binary>>}, State = #state{socket = Socket}) ->
|
||||
handle_info({tcp, Socket, <<?FRAME_REQUEST, _PacketId:32, MsgType:8, _/binary>>}, State = #state{socket = Socket}) ->
|
||||
logger:warning("[ws_channel] unsupported request message type: ~p", [MsgType]),
|
||||
{stop, State};
|
||||
|
||||
handle_info({tcp, Socket, <<?PACKET_CAST, ?MESSAGE_DATA, CastBin/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
handle_info({tcp, Socket, <<?FRAME_CAST, ?MESSAGE_DATA, CastBin/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
Data = message_pb:decode_msg(CastBin, 'Data'),
|
||||
iot_host:handle(HostPid, {data, Data}),
|
||||
{noreply, State};
|
||||
|
||||
handle_info({tcp, Socket, <<?PACKET_CAST, ?MESSAGE_EVENT_STREAM, CastBin/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
handle_info({tcp, Socket, <<?FRAME_CAST, ?MESSAGE_EVENT_STREAM, CastBin/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) ->
|
||||
CastMessage = message_pb:decode_msg(CastBin, 'TaskEventStream'),
|
||||
case CastMessage of
|
||||
#'TaskEventStream'{task_id = TaskId, type = Type0, stream = Reason0} when Type0 =:= <<"close">> ->
|
||||
@ -192,7 +192,7 @@ handle_info({tcp, Socket, <<?PACKET_CAST, ?MESSAGE_EVENT_STREAM, CastBin/binary>
|
||||
iot_event_stream_observer:stream_data(TaskId, Type, Stream)
|
||||
end,
|
||||
{noreply, State};
|
||||
handle_info({tcp, Socket, <<?PACKET_CAST, MsgType:8, _/binary>>}, State = #state{socket = Socket}) ->
|
||||
handle_info({tcp, Socket, <<?FRAME_CAST, MsgType:8, _/binary>>}, State = #state{socket = Socket}) ->
|
||||
logger:warning("[tcp_channel] unsupported cast message type: ~p", [MsgType]),
|
||||
{noreply, State};
|
||||
|
||||
@ -202,7 +202,7 @@ handle_info({tcp, Socket, <<?PACKET_CAST, MsgType:8, _/binary>>}, State = #state
|
||||
% {noreply, State};
|
||||
|
||||
%% 主机端的消息响应
|
||||
handle_info({tcp, Socket, <<?PACKET_RESPONSE, PacketId:32, ?MESSAGE_JSONRPC_REPLY, ReplyBin/binary>>}, State = #state{socket = Socket, inflight = Inflight}) when PacketId > 0 ->
|
||||
handle_info({tcp, Socket, <<?FRAME_RESPONSE, PacketId:32, ?MESSAGE_JSONRPC_REPLY, ReplyBin/binary>>}, State = #state{socket = Socket, inflight = Inflight}) when PacketId > 0 ->
|
||||
RpcReply = message_pb:decode_msg(ReplyBin, 'JsonRpcReply'),
|
||||
|
||||
case maps:take(PacketId, Inflight) of
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user