>>78412758>https://github.com/doubleyou/dps>cleanhandle_call({subscribe, Pid, TS}, _From, State = #state{messages = Messages, tag = Tag,
subscribers = Subscribers, last_ts = LastTS}) ->
T1 = erlang:now(),
Ref = erlang:monitor(process, Pid),
Msgs = messages_newer(Messages, TS),
case Msgs of
[] -> ok;
_ -> Pid ! {dps_msg, Tag, LastTS, Msgs}
end,
NewState = State#state{subscribers = [{Pid,Ref} | Subscribers]},
T2 = erlang:now(),
put(subscribe_count,get(subscribe_count)+1),
put(subscribe_time,get(subscribe_time)+timer:now_diff(T2,T1)),
{reply, length(Msgs), NewState};
handle_call({unsubscribe, Pid}, _From, State = #state{subscribers = Subscribers}) ->
T1 = erlang:now(),
{Delete, Remain} = lists:partition(fun({P,_Ref}) -> P == Pid end, Subscribers),
[erlang:demonitor(Ref) || {_Pid,Ref} <- Delete],
NewState = State#state{subscribers = Remain},
T2 = erlang:now(),
put(unsubscribe_count,get(unsubscribe_count)+1),
put(unsubscribe_time,get(unsubscribe_time)+timer:now_diff(T2,T1)),
{reply, ok, NewState};
handle_call({messages, TS}, _From, State = #state{last_ts = LastTS, messages = AllMessages}) ->
Messages = messages_newer(AllMessages, TS),
{reply, {ok, LastTS, Messages}, State};
handle_call({replication_messages, LastTS1, Msgs}, _From, State = #state{}) ->
{reply, ok, replication_messages(LastTS1, Msgs, State)};
handle_call(_Msg, _From, State) ->
{reply, {error, {unknown_call, _Msg}}, State}.
