yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

chore(gw-exproto): retry the closed grpc stream

This commit is contained in:
JianBo He 2021-10-13 09:55:15 +08:00
parent fc2e358f01
commit ac42a91fc2
2 changed files with 15 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
apps/emqx_gateway/src/exproto/emqx_exproto_channel.erl
View File

@ -271,7 +271,7 @@ handle_call({auth, ClientInfo0, Password}, _From,
conninfo = ConnInfo, conninfo = ConnInfo,
clientinfo = ClientInfo}) -> clientinfo = ClientInfo}) ->
ClientInfo1 = enrich_clientinfo(ClientInfo0, ClientInfo), ClientInfo1 = enrich_clientinfo(ClientInfo0, ClientInfo),
ConnInfo1 = enrich_conninfo(ClientInfo1, ConnInfo), ConnInfo1 = enrich_conninfo(ClientInfo0, ConnInfo),
Channel1 = Channel#channel{conninfo = ConnInfo1, Channel1 = Channel#channel{conninfo = ConnInfo1,
clientinfo = ClientInfo1}, clientinfo = ClientInfo1},
@ -383,7 +383,7 @@ handle_info({sock_closed, Reason},
andalso Inflight =:= undefined of andalso Inflight =:= undefined of
true -> true ->
Channel1 = ensure_disconnected({sock_closed, Reason}, Channel), Channel1 = ensure_disconnected({sock_closed, Reason}, Channel),
{shutdown, {sock_closed, Reason}, Channel1}; {shutdown, Reason, Channel1};
_ -> _ ->
%% delayed close process for flushing all callback funcs to gRPC server %% delayed close process for flushing all callback funcs to gRPC server
Channel1 = Channel#channel{closed_reason = {sock_closed, Reason}}, Channel1 = Channel#channel{closed_reason = {sock_closed, Reason}},

16
apps/emqx_gateway/src/exproto/emqx_exproto_gcli.erl
View File

@ -89,15 +89,25 @@ handle_cast({rpc, Fun, Req, Options, From}, State = #state{streams = Streams}) -
{ok, Stream} -> {ok, Stream} ->
case catch grpc_client:send(Stream, Req) of case catch grpc_client:send(Stream, Req) of
ok -> ok ->
?LOG(debug, "Send to ~p method successfully, request: ~0p", [Fun, Req]), ?LOG(debug, "Send to ~p method successfully, request: ~0p",
[Fun, Req]),
reply(From, Fun, ok), reply(From, Fun, ok),
{noreply, State#state{streams = Streams#{Fun => Stream}}}; {noreply, State#state{streams = Streams#{Fun => Stream}}};
{'EXIT', {not_found, _Stk}} ->
%% Not found the stream, reopen it
?LOG(info, "Can not find the old stream ref for ~s; "
"re-try with a new stream!", [Fun]),
handle_cast(
{rpc, Fun, Req, Options, From},
State#state{streams = maps:remove(Fun, Streams)});
{'EXIT', {timeout, _Stk}} -> {'EXIT', {timeout, _Stk}} ->
?LOG(error, "Send to ~p method timeout, request: ~0p", [Fun, Req]), ?LOG(error, "Send to ~p method timeout, request: ~0p",
[Fun, Req]),
reply(From, Fun, {error, timeout}), reply(From, Fun, {error, timeout}),
{noreply, State#state{streams = Streams#{Fun => Stream}}}; {noreply, State#state{streams = Streams#{Fun => Stream}}};
{'EXIT', {Reason1, _Stk}} -> {'EXIT', {Reason1, _Stk}} ->
?LOG(error, "Send to ~p method failure, request: ~0p, stacktrace: ~0p", [Fun, Req, _Stk]), ?LOG(error, "Send to ~p method failure, request: ~0p, "
"stacktrace: ~0p", [Fun, Req, _Stk]),
reply(From, Fun, {error, Reason1}), reply(From, Fun, {error, Reason1}),
{noreply, State#state{streams = Streams#{Fun => undefined}}} {noreply, State#state{streams = Streams#{Fun => undefined}}}
end end