Current section

Files

Jump to
kafe src kafe_rr.erl
Raw

src/kafe_rr.erl

% @hidden
-module(kafe_rr).
-behaviour(gen_server).
-define(SERVER, ?MODULE).
%% API.
-export([start_link/0]).
-export([next/1]).
-export([stop/0]).
%% gen_server.
-export([init/1]).
-export([handle_call/3]).
-export([handle_cast/2]).
-export([handle_info/2]).
-export([terminate/2]).
-export([code_change/3]).
%% API.
-spec start_link() -> {ok, pid()}.
start_link() ->
gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
stop() ->
gen_server:call(?SERVER, stop).
next(Topic) when is_binary(Topic) ->
gen_server:call(?SERVER, {next, Topic}).
%% gen_server.
init([]) ->
{ok, #{}}.
handle_call({next, Topic}, _From, State) ->
Partitions = case State of
#{Topic := P} -> P;
_ -> lists:reverse(kafe:partitions(Topic))
end,
case Partitions of
[] ->
{reply, 0, State};
_ ->
[Next|R] = lists:reverse(Partitions),
{reply, Next, State#{Topic => [Next|lists:reverse(R)]}}
end;
handle_call(stop, _From, State) ->
{stop, normal, shutdown_ok, State};
handle_call(_Request, _From, State) ->
{reply, ignored, State}.
handle_cast(_Msg, State) ->
{noreply, State}.
handle_info(_Info, State) ->
{noreply, State}.
terminate(_Reason, _State) ->
ok.
code_change(_OldVsn, State, _Extra) ->
{ok, State}.