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

refactor(sessds): make replay error handling a bit more clear 

Also leave a forgotten TODO.
This commit is contained in:
Andrew Mayorov 2024-03-05 20:26:18 +01:00
parent 3f3e33b2cb
commit b604c3dbd4
No known key found for this signature in database
GPG Key ID: 2837C62ACFBFED5D
1 changed files with 9 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
apps/emqx/src/emqx_persistent_session_ds.erl
View File

@ -532,8 +532,15 @@ replay_streams(Session0 = #{replay := [{_StreamKey, Srs0} | Rest]}, ClientInfo)
case replay_batch(Srs0, Session0, ClientInfo) of case replay_batch(Srs0, Session0, ClientInfo) of
Session = #{} -> Session = #{} ->
replay_streams(Session#{replay := Rest}, ClientInfo); replay_streams(Session#{replay := Rest}, ClientInfo);
{error, _, _} -> {error, recoverable, Reason} ->
?SLOG(warning, #{
msg => "failed_to_fetch_replay_batch",
stream => Srs0,
reason => Reason,
class => recoverable
}),
Session0 Session0
%% TODO: Handle unrecoverable errors.
end; end;
replay_streams(Session0 = #{replay := []}, _ClientInfo) -> replay_streams(Session0 = #{replay := []}, _ClientInfo) ->
Session = maps:remove(replay, Session0), Session = maps:remove(replay, Session0),
@ -554,13 +561,7 @@ replay_batch(Srs0, Session0, ClientInfo) ->
got => Srs got => Srs
}), }),
Session; Session;
{error, recoverable, Reason} = Error -> {error, _, _} = Error ->
?SLOG(warning, #{
msg => "failed_to_fetch_replay_batch",
stream => Srs0,
reason => Reason,
class => recoverable
}),
Error Error
end. end.