anlicheng/iot_cloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix

Browse Source
This commit is contained in:
anlicheng 2026-04-18 17:25:41 +08:00
parent 6c204d8c36
commit 5be57b1ae3
1 changed files with 11 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
src/transport/tcp/tcp_channel.erl
View File

@ -93,20 +93,17 @@ handle_call(_Request, _From, State) ->
%% 发送消息, 基于pub/sub机制 %% 发送消息, 基于pub/sub机制
handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, socket = Socket}) -> handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, socket = Socket}) ->
Encoded = message_pb:encode_msg(#'Pub'{topic = Topic, qos = Qos, content = Content}), Encoded = message_pb:encode_msg(#'Pub'{topic = Topic, qos = Qos, content = Content}),
EncPub = <<?MESSAGE_PUB, Encoded/binary>>, Transport:send(Socket, <<?PACKET_CAST, ?MESSAGE_PUB, Encoded/binary>>),
Transport:send(Socket, <<?PACKET_CAST, EncPub/binary>>),
{noreply, State}; {noreply, State};
%% 发送Command消息 %% 发送Command消息
handle_cast({command, CommandType, Command}, State = #state{transport = Transport, socket = Socket}) -> handle_cast({command, CommandType, Command}, State = #state{transport = Transport, socket = Socket}) ->
Encoded = message_pb:encode_msg(#'Command'{command_type = CommandType, command = Command}), Encoded = message_pb:encode_msg(#'Command'{command_type = CommandType, command = Command}),
EncCommand = <<?MESSAGE_COMMAND, Encoded/binary>>, Transport:send(Socket, <<?PACKET_CAST, ?MESSAGE_COMMAND, Encoded/binary>>),
Transport:send(Socket, <<?PACKET_CAST, EncCommand/binary>>),
{noreply, State}; {noreply, State};
%% 推送消息 %% 推送消息
handle_cast({jsonrpc_call, ReceiverPid, Ref, {Method, Params}}, State = #state{transport = Transport, socket = Socket, packet_id = PacketId, inflight = Inflight}) handle_cast({jsonrpc_call, ReceiverPid, Ref, {Method, Params}}, State = #state{transport = Transport, socket = Socket, packet_id = PacketId, inflight = Inflight}) ->
when is_binary(Method) ->
case next_packet_id(PacketId, Inflight) of case next_packet_id(PacketId, Inflight) of
{ok, NPacketId, NextPacketId} -> {ok, NPacketId, NextPacketId} ->
Encoded = message_pb:encode_msg(#'JsonRpcRequest'{ Encoded = message_pb:encode_msg(#'JsonRpcRequest'{
@ -117,10 +114,8 @@ handle_cast({jsonrpc_call, ReceiverPid, Ref, {Method, Params}}, State = #state{t
EncRequest = <<?MESSAGE_JSONRPC_REQUEST, Encoded/binary>>, EncRequest = <<?MESSAGE_JSONRPC_REQUEST, Encoded/binary>>,
TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {jsonrpc_timeout, NPacketId}), TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {jsonrpc_timeout, NPacketId}),
Transport:send(Socket, <<?PACKET_REQUEST, NPacketId:32, EncRequest/binary>>), Transport:send(Socket, <<?PACKET_REQUEST, NPacketId:32, EncRequest/binary>>),
{noreply, State#state{
packet_id = NextPacketId, {noreply, State#state{packet_id = NextPacketId, inflight = maps:put(NPacketId, {ReceiverPid, Ref, TimerRef}, Inflight)}};
inflight = maps:put(NPacketId, {ReceiverPid, Ref, TimerRef}, Inflight)
}};
{error, inflight_full} -> {error, inflight_full} ->
logger:warning("[ws_channel] uuid: ~p, inflight requests exhausted", [State#state.uuid]), logger:warning("[ws_channel] uuid: ~p, inflight requests exhausted", [State#state.uuid]),
{noreply, State} {noreply, State}
@ -176,22 +171,18 @@ handle_info({tcp, Socket, <<?PACKET_REQUEST, _PacketId:32, MsgType:8, _/binary>>
{stop, State}; {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, <<?PACKET_CAST, ?MESSAGE_DATA, CastBin/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) ->
CastMessage = message_pb:decode_msg(CastBin, 'Data'), Data = message_pb:decode_msg(CastBin, 'Data'),
case CastMessage of iot_host:handle(HostPid, {data, Data}),
#'Data'{} = Data ->
iot_host:handle(HostPid, {data, Data})
end,
{noreply, State}; {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, <<?PACKET_CAST, ?MESSAGE_EVENT_STREAM, CastBin/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) when is_pid(HostPid) ->
CastMessage = message_pb:decode_msg(CastBin, 'TaskEventStream'), CastMessage = message_pb:decode_msg(CastBin, 'TaskEventStream'),
case CastMessage of case CastMessage of
#'TaskEventStream'{task_id = TaskId, type = Type0, stream = Reason0} when Type0 =:= <<"close">>; Type0 =:= [<<"close">>] -> #'TaskEventStream'{task_id = TaskId, type = Type0, stream = Reason0} when Type0 =:= <<"close">> ->
iot_event_stream_observer:stream_close(TaskId, iolist_to_binary(Reason0)); iot_event_stream_observer:stream_close(TaskId, iolist_to_binary(Reason0));
#'TaskEventStream'{task_id = TaskId, type = Type, stream = Stream} -> #'TaskEventStream'{task_id = TaskId, type = Type, stream = Stream} ->
TypeBin = iolist_to_binary(Type), logger:debug("[tcp_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, Type, Stream]),
StreamBin = iolist_to_binary(Stream), iot_event_stream_observer:stream_data(TaskId, Type, Stream)
logger:debug("[tcp_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, TypeBin, StreamBin]),
iot_event_stream_observer:stream_data(TaskId, TypeBin, StreamBin)
end, end,
{noreply, State}; {noreply, State};
handle_info({tcp, Socket, <<?PACKET_CAST, MsgType:8, _/binary>>}, State = #state{socket = Socket}) -> handle_info({tcp, Socket, <<?PACKET_CAST, MsgType:8, _/binary>>}, State = #state{socket = Socket}) ->