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

Fix einval after socket was closed

This commit is contained in:
zhouzb 2020-03-05 10:25:15 +08:00 committed by turtleDeng
parent caed16f57b
commit ad7cbb4428
2 changed files with 21 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/emqx_channel.erl
View File

@ -808,6 +808,9 @@ handle_info(Info, Channel) ->
handle_timeout(_TRef, {keepalive, _StatVal}, handle_timeout(_TRef, {keepalive, _StatVal},
Channel = #channel{keepalive = undefined}) -> Channel = #channel{keepalive = undefined}) ->
{ok, Channel}; {ok, Channel};
handle_timeout(_TRef, {keepalive, _StatVal},
Channel = #channel{conn_state = disconnected}) ->
{ok, Channel};
handle_timeout(_TRef, {keepalive, StatVal}, handle_timeout(_TRef, {keepalive, StatVal},
Channel = #channel{keepalive = Keepalive}) -> Channel = #channel{keepalive = Keepalive}) ->
case emqx_keepalive:check(StatVal, Keepalive) of case emqx_keepalive:check(StatVal, Keepalive) of
@ -818,6 +821,9 @@ handle_timeout(_TRef, {keepalive, StatVal},
handle_out(disconnect, ?RC_KEEP_ALIVE_TIMEOUT, Channel) handle_out(disconnect, ?RC_KEEP_ALIVE_TIMEOUT, Channel)
end; end;
handle_timeout(_TRef, retry_delivery,
Channel = #channel{conn_state = disconnected}) ->
{ok, Channel};
handle_timeout(_TRef, retry_delivery, handle_timeout(_TRef, retry_delivery,
Channel = #channel{session = Session}) -> Channel = #channel{session = Session}) ->
case emqx_session:retry(Session) of case emqx_session:retry(Session) of
@ -831,6 +837,9 @@ handle_timeout(_TRef, retry_delivery,
handle_out(publish, Publishes, reset_timer(retry_timer, Timeout, NChannel)) handle_out(publish, Publishes, reset_timer(retry_timer, Timeout, NChannel))
end; end;
handle_timeout(_TRef, expire_awaiting_rel,
Channel = #channel{conn_state = disconnected}) ->
{ok, Channel};
handle_timeout(_TRef, expire_awaiting_rel, handle_timeout(_TRef, expire_awaiting_rel,
Channel = #channel{session = Session}) -> Channel = #channel{session = Session}) ->
case emqx_session:expire(awaiting_rel, Session) of case emqx_session:expire(awaiting_rel, Session) of

19
src/emqx_connection.erl
View File

@ -476,13 +476,18 @@ handle_timeout(_TRef, emit_stats, State =
emqx_cm:set_chan_stats(ClientId, stats(State)), emqx_cm:set_chan_stats(ClientId, stats(State)),
{ok, State#state{stats_timer = undefined}}; {ok, State#state{stats_timer = undefined}};
handle_timeout(TRef, keepalive, State = handle_timeout(TRef, keepalive, State = #state{transport = Transport,
#state{transport = Transport, socket = Socket}) -> socket = Socket,
case Transport:getstat(Socket, [recv_oct]) of channel = Channel})->
{ok, [{recv_oct, RecvOct}]} -> case emqx_channel:info(conn_state, Channel) of
handle_timeout(TRef, {keepalive, RecvOct}, State); disconnected -> {ok, State};
{error, Reason} -> _ ->
handle_info({sock_error, Reason}, State) case Transport:getstat(Socket, [recv_oct]) of
{ok, [{recv_oct, RecvOct}]} ->
handle_timeout(TRef, {keepalive, RecvOct}, State);
{error, Reason} ->
handle_info({sock_error, Reason}, State)
end
end; end;
handle_timeout(TRef, Msg, State) -> handle_timeout(TRef, Msg, State) ->