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

fix rpc

Browse Source
This commit is contained in:
anlicheng 2026-04-19 17:36:55 +08:00
parent b3d795c523
commit 5f75189c7a
2 changed files with 12 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
src/host/iot_host.erl
View File

@ -135,7 +135,7 @@ await_reply(Pid, Ref, Timeout) when is_pid(Pid), is_reference(Ref), is_integer(T
{rpc_reply, Ref, #'RpcReply'{reply = {error, #'RpcReply.RpcError'{code = Code, message = Message}}}} -> {rpc_reply, Ref, #'RpcReply'{reply = {error, #'RpcReply.RpcError'{code = Code, message = Message}}}} ->
{error, Code, unicode:characters_to_binary(Message)} {error, Code, unicode:characters_to_binary(Message)}
after Timeout -> after Timeout ->
ok = gen_statem:call(Pid, {cancel_jsonrpc_call, Ref}), ok = gen_statem:call(Pid, {cancel_rpc_call, Ref}),
flush_reply(Ref), flush_reply(Ref),
{error, -1, <<"timeout">>} {error, -1, <<"timeout">>}
end. end.
@ -215,11 +215,11 @@ handle_event({call, From}, get_status, _, State = #state{channel_pid = ChannelPi
{keep_state, State, [{reply, From, {ok, Reply}}]}; {keep_state, State, [{reply, From, {ok, Reply}}]};
%% 只要channel存在，就负责将消息推送到边缘端主机 %% 只要channel存在，就负责将消息推送到边缘端主机
handle_event({call, From}, {jsonrpc_call, ReceiverPid, RpcCall}, _, State = #state{uuid = UUID, channel_pid = ChannelPid, has_session = HasSession}) -> handle_event({call, From}, {rpc_call, ReceiverPid, RpcCall}, _, State = #state{uuid = UUID, channel_pid = ChannelPid, has_session = HasSession}) ->
case HasSession andalso is_pid(ChannelPid) of case HasSession andalso is_pid(ChannelPid) of
true -> true ->
%% 通过websocket发送请求 %% 通过websocket发送请求
Ref = tcp_channel:jsonrpc_call(ChannelPid, ReceiverPid, RpcCall), Ref = tcp_channel:rpc_call(ChannelPid, ReceiverPid, RpcCall),
{keep_state, State, [{reply, From, {ok, Ref}}]}; {keep_state, State, [{reply, From, {ok, Ref}}]};
false -> false ->
logger:debug("[iot_host] uuid: ~p, invalid state: ~p", [UUID, state_map(State)]), logger:debug("[iot_host] uuid: ~p, invalid state: ~p", [UUID, state_map(State)]),
@ -236,10 +236,10 @@ handle_event({call, From}, {container_call, ReceiverPid, Request}, _, State = #s
{keep_state, State, [{reply, From, {error, <<"主机离线，发送请求失败"/utf8>>}}]} {keep_state, State, [{reply, From, {error, <<"主机离线，发送请求失败"/utf8>>}}]}
end; end;
handle_event({call, From}, {cancel_jsonrpc_call, Ref}, _, State = #state{channel_pid = ChannelPid}) -> handle_event({call, From}, {cancel_rpc_call, Ref}, _, State = #state{channel_pid = ChannelPid}) ->
case is_pid(ChannelPid) of case is_pid(ChannelPid) of
true -> true ->
ok = tcp_channel:cancel_jsonrpc_call(ChannelPid, Ref), ok = tcp_channel:cancel_rpc_call(ChannelPid, Ref),
{keep_state, State, [{reply, From, ok}]}; {keep_state, State, [{reply, From, ok}]};
false -> false ->
{keep_state, State, [{reply, From, ok}]} {keep_state, State, [{reply, From, ok}]}

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

@ -16,7 +16,7 @@
-define(INFLIGHT_TIMEOUT, 60000). -define(INFLIGHT_TIMEOUT, 60000).
%% API %% API
-export([pub/4, jsonrpc_call/3, container_call/3, cancel_jsonrpc_call/2, command/3]). -export([pub/4, rpc_call/3, container_call/3, cancel_rpc_call/2, command/3]).
-export([start_link/3, stop/2]). -export([start_link/3, stop/2]).
%% gen_server callbacks %% gen_server callbacks
@ -53,8 +53,8 @@ command(Pid, CommandType, Command) when is_pid(Pid), is_integer(CommandType), is
gen_server:cast(Pid, {command, CommandType, Command}). gen_server:cast(Pid, {command, CommandType, Command}).
%% 向通道中写入消息 %% 向通道中写入消息
-spec jsonrpc_call(Pid :: pid(), ReceiverPid :: pid(), Request :: {Method :: binary(), Params :: binary()}) -> Ref :: reference(). -spec rpc_call(Pid :: pid(), ReceiverPid :: pid(), Request :: {Method :: binary(), Params :: binary()}) -> Ref :: reference().
jsonrpc_call(Pid, ReceiverPid, {Method, Params}) when is_pid(Pid), is_pid(ReceiverPid), is_binary(Method), is_binary(Params) -> rpc_call(Pid, ReceiverPid, {Method, Params}) when is_pid(Pid), is_pid(ReceiverPid), is_binary(Method), is_binary(Params) ->
Ref = make_ref(), Ref = make_ref(),
RpcRequest = #'RpcRequest'{method = Method, params = Params}, RpcRequest = #'RpcRequest'{method = Method, params = Params},
gen_server:cast(Pid, {request_call, ReceiverPid, Ref, {rpc_request, RpcRequest}}), gen_server:cast(Pid, {request_call, ReceiverPid, Ref, {rpc_request, RpcRequest}}),
@ -66,9 +66,9 @@ container_call(Pid, ReceiverPid, Request) when is_pid(Pid), is_pid(ReceiverPid),
gen_server:cast(Pid, {request_call, ReceiverPid, Ref, {container_request, Request}}), gen_server:cast(Pid, {request_call, ReceiverPid, Ref, {container_request, Request}}),
Ref. Ref.
-spec cancel_jsonrpc_call(Pid :: pid(), Ref :: reference()) -> ok. -spec cancel_rpc_call(Pid :: pid(), Ref :: reference()) -> ok.
cancel_jsonrpc_call(Pid, Ref) when is_pid(Pid), is_reference(Ref) -> cancel_rpc_call(Pid, Ref) when is_pid(Pid), is_reference(Ref) ->
gen_server:call(Pid, {cancel_jsonrpc_call, Ref}). gen_server:call(Pid, {cancel_rpc_call, Ref}).
%% 关闭方法 %% 关闭方法
-spec stop(Pid :: pid(), Reason :: any()) -> no_return(). -spec stop(Pid :: pid(), Reason :: any()) -> no_return().
@ -92,7 +92,7 @@ init(Ref, Transport, _Opts = []) ->
% erlang:start_timer(?PING_TICKER, self(), ping_ticker), % erlang:start_timer(?PING_TICKER, self(), ping_ticker),
gen_server:enter_loop(?MODULE, [], #state{transport = Transport, socket = Socket}). gen_server:enter_loop(?MODULE, [], #state{transport = Transport, socket = Socket}).
handle_call({cancel_jsonrpc_call, Ref}, _From, State = #state{inflight = Inflight}) -> handle_call({cancel_rpc_call, Ref}, _From, State = #state{inflight = Inflight}) ->
case take_inflight_by_ref(Ref, Inflight) of case take_inflight_by_ref(Ref, Inflight) of
{ok, #inflight_request{timer_ref = TimerRef}, NInflight} -> {ok, #inflight_request{timer_ref = TimerRef}, NInflight} ->
erlang:cancel_timer(TimerRef), erlang:cancel_timer(TimerRef),