diff --git a/src/host/iot_host.erl b/src/host/iot_host.erl index 89ae9b6..1c9e5fd 100644 --- a/src/host/iot_host.erl +++ b/src/host/iot_host.erl @@ -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}}}} -> {error, Code, unicode:characters_to_binary(Message)} after Timeout -> - ok = gen_statem:call(Pid, {cancel_jsonrpc_call, Ref}), + ok = gen_statem:call(Pid, {cancel_rpc_call, Ref}), flush_reply(Ref), {error, -1, <<"timeout">>} end. @@ -215,11 +215,11 @@ handle_event({call, From}, get_status, _, State = #state{channel_pid = ChannelPi {keep_state, State, [{reply, From, {ok, Reply}}]}; %% 只要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 true -> %% 通过websocket发送请求 - Ref = tcp_channel:jsonrpc_call(ChannelPid, ReceiverPid, RpcCall), + Ref = tcp_channel:rpc_call(ChannelPid, ReceiverPid, RpcCall), {keep_state, State, [{reply, From, {ok, Ref}}]}; false -> 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>>}}]} 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 true -> - ok = tcp_channel:cancel_jsonrpc_call(ChannelPid, Ref), + ok = tcp_channel:cancel_rpc_call(ChannelPid, Ref), {keep_state, State, [{reply, From, ok}]}; false -> {keep_state, State, [{reply, From, ok}]} diff --git a/src/transport/tcp/tcp_channel.erl b/src/transport/tcp/tcp_channel.erl index 47db5e7..be00d67 100644 --- a/src/transport/tcp/tcp_channel.erl +++ b/src/transport/tcp/tcp_channel.erl @@ -16,7 +16,7 @@ -define(INFLIGHT_TIMEOUT, 60000). %% 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]). %% 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}). %% 向通道中写入消息 --spec jsonrpc_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) -> +-spec rpc_call(Pid :: pid(), ReceiverPid :: pid(), Request :: {Method :: binary(), Params :: binary()}) -> Ref :: reference(). +rpc_call(Pid, ReceiverPid, {Method, Params}) when is_pid(Pid), is_pid(ReceiverPid), is_binary(Method), is_binary(Params) -> Ref = make_ref(), RpcRequest = #'RpcRequest'{method = Method, params = Params}, 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}}), Ref. --spec cancel_jsonrpc_call(Pid :: pid(), Ref :: reference()) -> ok. -cancel_jsonrpc_call(Pid, Ref) when is_pid(Pid), is_reference(Ref) -> - gen_server:call(Pid, {cancel_jsonrpc_call, Ref}). +-spec cancel_rpc_call(Pid :: pid(), Ref :: reference()) -> ok. +cancel_rpc_call(Pid, Ref) when is_pid(Pid), is_reference(Ref) -> + gen_server:call(Pid, {cancel_rpc_call, Ref}). %% 关闭方法 -spec stop(Pid :: pid(), Reason :: any()) -> no_return(). @@ -92,7 +92,7 @@ init(Ref, Transport, _Opts = []) -> % erlang:start_timer(?PING_TICKER, self(), ping_ticker), 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 {ok, #inflight_request{timer_ref = TimerRef}, NInflight} -> erlang:cancel_timer(TimerRef),