anchor.diff:  stop(Pid) ->
     gen_server:call(Pid, stop, infinity).
   
  +%% @doc Subscribe on messages from a partition
   -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.
  +%% @doc Unsubscribe the current subscriber.
   -spec unsubscribe(pid()) -> ok.
   unsubscribe(Pid) ->
     gen_server:call(Pid, unsubscribe).
  
positive.code:   {noreply, State}.
  
  handle_call({subscribe, Pid, Options}, _From,
              #state{subscriber = Subscriber} = State0) ->
    case Subscriber =:= undefined orelse
        not is_process_alive(Subscriber) orelse
        Subscriber =:= Pid of
      true ->
        case update_options(Options, State0) of
          {ok, State1} ->
            Mref = erlang:monitor(process, Pid),
            State2 = State1#state{ subscriber      = Pid
                                 , subscriber_mref = Mref
  
negative.code:     []       -> {error, no_available_offsets}
    end.
  
  %% @private Reset fetch buffer, use the last unacked offset as the next begin
  %% offset to fetch data from.
  %% @end
  -spec reset_buffer(#state{}) -> #state{}.
  