anchor.diff:      []       -> {error, no_available_offsets}
     end.
   
  -%% @private Rest fetch buffer, use the last unacked offset as the next begin
  +%% @private Reset fetch buffer, use the last unacked offset as the next begin
   %% offset to fetch data from.
   %% @end
   -spec reset_buffer(#state{}) -> #state{}.
  
positive.code: stop(Pid) ->
    gen_server:call(Pid, stop, infinity).
  
  -spec subscribe(pid(), pid(), options()) -> ok | {error, any()}.
  subscribe(Pid, SubscriberPid, ConsumerOptions) ->
    gen_server:call(Pid, {subscribe, SubscriberPid, ConsumerOptions}, infinity).
  
  %% @doc Ubsubscribe the current subscriber.
  -spec unsubscribe(pid()) -> ok.
  unsubscribe(Pid) ->
    gen_server:call(Pid, unsubscribe).
  
negative.code:   {noreply, State}.
  
  handle_call({subscribe, Pid, Options}, _From,
              #state{ subscriber = Subscriber
                    , subscriber_mref = OldMref} = State0) ->
    case Subscriber =:= undefined orelse
        not is_process_alive(Subscriber) orelse
        Subscriber =:= Pid of
      true ->
        case update_options(Options, State0) of
          {ok, State1} ->
            %% demonitor in case the same process tries to subscribe again
            is_reference(OldMref) andalso erlang:demonitor(OldMref, [flush]),
            Mref = erlang:monitor(process, Pid),
            State2 = State1#state{ subscriber      = Pid
                                 , subscriber_mref = Mref
  