anchor.diff:           handle_coverage/4,
            handle_exit/3]).
   
  --record(state, {partition}).
  +-record(state, {partition, channels}).
   
   %% API
   start_vnode(I) ->
       riak_core_vnode_master:get_vnode_pid(I, ?MODULE).
   
   init([Partition]) ->
  -    {ok, #state{partition=Partition}}.
  +    {ok, #state{partition=Partition, channels=dict:new()}}.
   
   %% @doc Respond to a subscription; launch a child process,
   %%      and register it with gproc under a given channel name.
  -handle_command({subscribe, Channel, Pid}, _Sender, State) ->
  +handle_command({subscribe, Channel, Pid},
  +               _Sender,
  +               #state{channels=Channels0}=State) ->
       lager:warning("Received subscribe for ~p and ~p.\n",
                     [Channel, Pid]),
  -    riak_pubsub_subscription_sup:start_child(Channel, Pid),
  -    {reply, ok, State};
  +
  +    case subscribe(Channels0, Channel, Pid) of
  +        {error, Error} ->
  +            {reply, {error, Error}, State};
  +        {ok, Channels} ->
  +            {reply, ok, State#state{channels=Channels}}
  +    end;
   handle_command(Message, _Sender, State) ->
       ?PRINT({unhandled_command, Message}),
       {noreply, State}.
   
  -handle_handoff_command(_Message, _Sender, State) ->
  -    {noreply, State}.
  +handle_handoff_command(?FOLD_REQ{foldfun=Fun, acc0=Acc0}, _Sender, State) ->
  +    Acc = dict:fold(Fun, Acc0, State#state.channels),
  +    {reply, Acc, State}.
   
   handoff_starting(_TargetNode, State) ->
       {true, State}.
  
positive.code: 
  terminate(_Reason, _State) ->
      ok.
  
negative.code: handoff_finished(_TargetNode, State) ->
      {ok, State}.
  
  handle_handoff_data(Data, #state{channels=Channels0}=State) ->
      {Channel, Pids} = binary_to_term(Data),
  
      case subscribe(Channels0, Channel, Pids) of
          {error, Error} ->
              {reply, {error, Error}, State};
          {ok, Channels} ->
              {reply, ok, State#state{channels=Channels}}
      end.
  
  encode_handoff_item(Channel, Pids) ->
      term_to_binary({Channel, Pids}).
  
  is_empty(#state{channels=Channels}=State) ->
      case dict:size(Channels) of
          0 -> {true, State};
          _ -> {false, State}
      end.
  
  delete(State) ->
      {ok, State}.
  