Packages
kafka_protocol
4.1.8
4.3.4
4.3.3
4.3.2
4.3.1
4.3.0
4.2.9
4.2.8
4.2.7
4.2.6
4.2.5
4.2.4
4.2.3
4.2.2
4.2.1
4.2.0
4.1.10
4.1.9
4.1.8
4.1.7
4.1.6
4.1.5
4.1.4
4.1.3
4.1.2
4.1.1
4.1.0
4.0.3
4.0.2
4.0.1
3.0.1
3.0.0
2.4.1
2.3.6
2.3.5
2.3.4
2.3.3
2.3.2
2.3.1
2.3.0
2.2.9
2.2.8
2.2.7
2.2.6
2.2.5
2.2.4
2.2.3
2.2.2
2.2.1
2.2.0
2.1.2
2.1.1
2.1.0
2.0.1
2.0.0
1.1.3
1.1.2
1.1.1
1.1.0
1.0.0
0.9.2
0.9.1
0.9.0
0.8.0
0.7.8
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.0
0.4.0
0.3.2
0.3.1
0.2.3
Kafka protocol library for Erlang/Elixir
Current section
Files
Jump to
Current section
Files
src/kpro_sent_reqs.erl
%%%
%%% Copyright (c) 2014-2021, Klarna Bank AB (publ)
%%%
%%% Licensed under the Apache License, Version 2.0 (the "License");
%%% you may not use this file except in compliance with the License.
%%% You may obtain a copy of the License at
%%%
%%% http://www.apache.org/licenses/LICENSE-2.0
%%%
%%% Unless required by applicable law or agreed to in writing, software
%%% distributed under the License is distributed on an "AS IS" BASIS,
%%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%%% See the License for the specific language governing permissions and
%%% limitations under the License.
%%%
%%%=============================================================================
%%% @doc
%%% This module manages an opaque collection of sent-requests.
%%% @end
%%% ============================================================================
%% @private
-module(kpro_sent_reqs).
%%%_* Exports ==================================================================
%% API
-export([ new/0
, add/5
, del/2
, get_req/2
, get_corr_id/1
, increment_corr_id/1
, scan_for_max_age/1
, is_empty/1
]).
-export_type([requests/0]).
-type corr_id() :: kpro:corr_id().
-define(REQ(Caller, Ref, API, Vsn, Ts), {Caller, Ref, API, Vsn, Ts}).
-type req() :: ?REQ(pid(), reference(), kpro:api(), kpro:vsn(), integer()).
-record(requests,
{ corr_id = 0
, sent = #{} :: #{corr_id() => req()}
}).
-opaque requests() :: #requests{}.
-include("kpro_private.hrl").
%%%_* APIs =====================================================================
-spec new() -> requests().
new() -> #requests{}.
-spec is_empty(requests()) -> boolean().
is_empty(#requests{sent = Sent}) -> maps:size(Sent) == 0.
%% @doc Add a new request to sent collection.
%% Return the last corrlation ID and the new collection.
-spec add(requests(), pid(), reference(), kpro:api(), kpro:vsn()) ->
{corr_id(), requests()}.
add(#requests{ corr_id = CorrId
, sent = Sent
} = Requests, Caller, Ref, API, Vsn) ->
Req = ?REQ(Caller, Ref, API, Vsn, now_ts()),
NewSent = maps:put(CorrId, Req, Sent),
NewRequests = Requests#requests{ corr_id = next_corr_id(CorrId)
, sent = NewSent
},
{CorrId, NewRequests}.
%% @doc Delete a request from the opaque collection.
%% Crash if correlation ID is not found.
-spec del(requests(), corr_id()) -> requests().
del(#requests{sent = Sent} = Requests, CorrId) ->
Requests#requests{sent = maps:remove(CorrId, Sent)}.
%% @doc Get caller of a request having the given correlation ID.
%% Crash if the request is not found.
-spec get_req(requests(), corr_id()) ->
{pid(), reference(), kpro:api(), kpro:vsn()}.
get_req(#requests{sent = Sent}, CorrId) ->
?REQ(Caller, Ref, API, Vsn, _Ts) = maps:get(CorrId, Sent),
{Caller, Ref, API, Vsn}.
%% @doc Get the correction to be sent for the next request.
-spec get_corr_id(requests()) -> corr_id().
get_corr_id(#requests{ corr_id = CorrId }) ->
CorrId.
%% @doc Fetch and increment the correlation ID
%% This is used if we don't want a response from the broker
-spec increment_corr_id(requests()) -> {corr_id(), requests()}.
increment_corr_id(#requests{corr_id = CorrId} = Requests) ->
{CorrId, Requests#requests{ corr_id = next_corr_id(CorrId) }}.
%% @doc Scan all sent requests to get oldest sent request.
%% Age is in milli-seconds.
%% 0 is returned if there is no pending response.
-spec scan_for_max_age(requests()) -> timeout().
scan_for_max_age(#requests{sent = Sent}) ->
Now = now_ts(),
MinTs = maps:fold(fun(_CorrId, ?REQ(_Caller, _Ref, _API, _Vsn, Ts), Min) ->
erlang:min(Ts, Min)
end, Now, Sent),
Now - MinTs.
next_corr_id(N) -> (N + 1) rem (1 bsl 31).
%% os:system_time('millisecond') is since otp-19
now_ts() -> os:system_time(?millisecond).
%%%_* Emacs ====================================================================
%%% Local Variables:
%%% allout-layout: t
%%% erlang-indent-level: 2
%%% End: