anchor.diff:  -behaviour(riak_core_vnode).
   
   -include_lib("riak_pubsub.hrl").
  +-include_lib("riak_core/include/riak_core_vnode.hrl").
   
   -export([start_vnode/1,
            init/1,
  
positive.code:          handle_coverage/4,
           handle_exit/3]).
  
  -record(state, {partition}).
  
  %% API
  start_vnode(I) ->
      riak_core_vnode_master:get_vnode_pid(I, ?MODULE).
  
  init([Partition]) ->
      {ok, #state{partition=Partition}}.
  
  %% @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) ->
      lager:warning("Received subscribe for ~p and ~p.\n",
                    [Channel, Pid]),
      riak_pubsub_subscription_sup:start_child(Channel, Pid),
      {reply, ok, State};
  handle_command(Message, _Sender, State) ->
      ?PRINT({unhandled_command, Message}),
      {noreply, State}.
  
  handle_handoff_command(_Message, _Sender, State) ->
      {noreply, State}.
  
  handoff_starting(_TargetNode, State) ->
      {true, State}.
  
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}.
  