diff --git a/src/adapters/http/http_client.erl b/src/adapters/http/http_client.erl deleted file mode 100644 index 38286d8..0000000 --- a/src/adapters/http/http_client.erl +++ /dev/null @@ -1,34 +0,0 @@ -%%%------------------------------------------------------------------- -%%% @author anlicheng -%%% @copyright (C) 2025, -%%% @doc -%%% -%%% @end -%%% Created : 18. 8月 2025 18:40 -%%%------------------------------------------------------------------- --module(http_client). --author("anlicheng"). - -%% API --export([post/3]). - -%% Headers = [ -%% {<<"content-type">>, <<"application/json">>} -%% ] --spec post(Url :: string(), Headers :: list(), Body :: binary()) -> {ok, RespBody :: binary()} | {error, Reason :: any()}. -post(Url, Headers, Body) when is_list(Url), is_list(Headers), is_binary(Body) -> - case hackney:request(post, Url, Headers, Body, [{pool, false}]) of - {ok, 200, _, ClientRef} -> - {ok, RespBody} = hackney:body(ClientRef), - logger:debug("[http_client] send body: ~p, get error is: ~p", [Body, RespBody]), - hackney:close(ClientRef), - {ok, RespBody}; - {ok, HttpCode, _, ClientRef} -> - {ok, RespBody} = hackney:body(ClientRef), - hackney:close(ClientRef), - logger:warning("[http_client] send body: ~p, get error is: ~p", [Body, {HttpCode, RespBody}]), - {error, {HttpCode, RespBody}}; - {error, Reason} -> - logger:warning("[http_client] send body: ~p, get error is: ~p", [Body, Reason]), - {error, Reason} - end. diff --git a/src/adapters/http/iot_http_client.erl b/src/adapters/http/iot_http_client.erl deleted file mode 100644 index d4e72d9..0000000 --- a/src/adapters/http/iot_http_client.erl +++ /dev/null @@ -1,40 +0,0 @@ -%%%------------------------------------------------------------------- -%%% @author licheng5 -%%% @copyright (C) 2023, -%%% @doc -%%% -%%% @end -%%% Created : 03. 3月 2023 11:48 -%%%------------------------------------------------------------------- --module(iot_http_client). --author("licheng5"). - -%% API --export([post/2]). - -post(Url, Body) when is_list(Url), is_binary(Body) -> - case hackney:request(post, Url, [], Body) of - {ok, 200, _, ClientRef} -> - case hackney:body(ClientRef) of - {ok, RespBody} -> - logger:debug("[iot_http_client] url: ~p, response is: ~p", [Url, RespBody]), - ok; - {error, Reason} -> - logger:warning("[iot_http_client] url: ~p, get error: ~p", [Url, Reason]), - {error, Reason} - end; - - {ok, HttpCode, _, ClientRef} -> - case hackney:body(ClientRef) of - {ok, RespBody} -> - logger:debug("[iot_http_client] url: ~p, http_code: ~p, response is: ~p", [Url, HttpCode, RespBody]), - ok; - {error, Reason} -> - logger:warning("[iot_http_client] url: ~p, http_code: ~p, get error: ~p", [Url, HttpCode, Reason]), - {error, Reason} - end; - - {error, Reason} -> - logger:warning("[iot_http_client] url: ~p, get error: ~p", [Url, Reason]), - {error, Reason} - end. \ No newline at end of file diff --git a/src/transport/tcp/tcp_channel.erl b/src/transport/tcp/tcp_channel.erl index 7d77305..47db5e7 100644 --- a/src/transport/tcp/tcp_channel.erl +++ b/src/transport/tcp/tcp_channel.erl @@ -121,21 +121,18 @@ handle_cast({command, CommandType, Command}, State = #state{transport = Transpor %% 推送需要响应的请求 handle_cast({request_call, ReceiverPid, Ref, Body}, State = #state{transport = Transport, socket = Socket, packet_id = PacketId, inflight = Inflight}) -> - case next_packet_id(PacketId, Inflight) of - {ok, NPacketId, NextPacketId} -> - Encoded = message_pb:encode_msg(#'RequestFrame'{ - packet_id = NPacketId, - body = Body - }), - TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {request_timeout, NPacketId}), - Transport:send(Socket, <>), + Encoded = message_pb:encode_msg(#'RequestFrame'{ + packet_id = PacketId, + body = Body + }), + TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {request_timeout, PacketId}), + Transport:send(Socket, <>), - RequestInfo = #inflight_request{receiver_pid = ReceiverPid, ref = Ref, timer_ref = TimerRef}, - {noreply, State#state{packet_id = NextPacketId, inflight = maps:put(NPacketId, RequestInfo, Inflight)}}; - {error, inflight_full} -> - logger:warning("[ws_channel] uuid: ~p, inflight requests exhausted", [State#state.uuid]), - {noreply, State} - end. + RequestInfo = #inflight_request{receiver_pid = ReceiverPid, ref = Ref, timer_ref = TimerRef}, + {noreply, State#state{ + packet_id = inc_packet_id(PacketId), + inflight = maps:put(PacketId, RequestInfo, Inflight) + }}. %% auth验证 handle_info({tcp, Socket, <>}, State = #state{transport = Transport, socket = Socket}) -> @@ -205,42 +202,17 @@ take_inflight_by_ref(Ref, Inflight) -> end end, error, Inflight). --spec next_packet_id(integer(), map()) -> - {ok, integer(), integer()} | {error, inflight_full}. -next_packet_id(_PacketId, Inflight) when map_size(Inflight) >= ?MAX_PACKET_ID -> - {error, inflight_full}; -next_packet_id(PacketId, Inflight) -> - next_packet_id(PacketId, Inflight, PacketId, 0). - --spec next_packet_id(integer(), map(), integer(), non_neg_integer()) -> - {ok, integer(), integer()} | {error, inflight_full}. -next_packet_id(_PacketId, _Inflight, _StartPacketId, TryCount) when TryCount > ?MAX_PACKET_ID -> - {error, inflight_full}; -next_packet_id(PacketId, Inflight, StartPacketId, TryCount) -> - case maps:is_key(PacketId, Inflight) of - false -> - {ok, PacketId, inc_packet_id(PacketId)}; - true -> - NPacketId = inc_packet_id(PacketId), - case NPacketId =:= StartPacketId of - true -> - {error, inflight_full}; - false -> - next_packet_id(NPacketId, Inflight, StartPacketId, TryCount + 1) - end - end. - -spec inc_packet_id(integer()) -> integer(). inc_packet_id(?MAX_PACKET_ID) -> 1; inc_packet_id(PacketId) when PacketId > 0, PacketId < ?MAX_PACKET_ID -> PacketId + 1. --spec handle_request_frame(message_pb:'RequestFrame'(), module(), any(), #state{}) -> - {noreply, #state{}} | {stop, #state{}}. +-spec handle_request_frame(message_pb:'RequestFrame'(), module(), any(), #state{}) -> {noreply, #state{}} | {stop, #state{}}. handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = {auth_request, #'AuthRequest'{uuid = UUID, username = Username, token = Token, salt = Salt, timestamp = Timestamp}}}, Transport, Socket, State) -> + logger:debug("[ws_channel] auth uuid: ~p", [UUID]), case iot_auth:check(Username, Token, UUID, Salt, Timestamp) of true -> @@ -310,14 +282,14 @@ handle_response_frame(#'ResponseFrame'{packet_id = PacketId, body = {rpc_reply, case maps:take(PacketId, Inflight) of error -> {noreply, State}; - {#inflight_request{receiver_pid = ReceiverPid, ref = Ref, timer_ref = TimerRef}, NInflight} -> - erlang:cancel_timer(TimerRef), - case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of - true -> - ReceiverPid ! {rpc_reply, Ref, RpcReply}; - false -> - logger:warning("[ws_channel] get async_call_reply message: ~p, packet_id: ~p, but receiver_pid is deaded", [RpcReply, PacketId]) - end, + {#inflight_request{receiver_pid = ReceiverPid, ref = Ref, timer_ref = TimerRef}, NInflight} -> + erlang:cancel_timer(TimerRef), + case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of + true -> + ReceiverPid ! {rpc_reply, Ref, RpcReply}; + false -> + logger:warning("[ws_channel] get async_call_reply message: ~p, packet_id: ~p, but receiver_pid is deaded", [RpcReply, PacketId]) + end, {noreply, State#state{inflight = NInflight}} end; handle_response_frame(#'ResponseFrame'{packet_id = PacketId, body = {auth_reply, AuthReply}}, _Inflight, State) ->