anchor.diff:                  tcp_mod  = if SslOpts /= [] -> ssl;
                                 true          -> gen_tcp
                              end,
  -                timeout_len = app_helper:get_env(riak_core, handoff_receive_timeout, ?RECV_TIMEOUT)}}.
  +                recv_timeout_len = app_helper:get_env(riak_core, handoff_receive_timeout, ?RECV_TIMEOUT),
  +                vnode_timeout_len = app_helper:get_env(riak_core, handoff_receive_vnode_timeout, ?VNODE_TIMEOUT)}}.
   
   handle_call({set_socket, Socket0}, _From, State = #state{ssl_opts = SslOpts}) ->
       SockOpts = [{active, once}, {packet, 4}, {header, 1}],
  
positive.code:     State#state{partition=Partition, vnode=VNode};
  
  process_message(?PT_MSG_BATCH, MsgData, State) ->
      lists:foldl(fun(Obj, StateAcc) -> process_message(?PT_MSG_OBJ, Obj, StateAcc) end, 
                  State,
                  binary_to_term(MsgData));
  
  process_message(?PT_MSG_OBJ, MsgData, State=#state{vnode=VNode, count=Count}) ->
      Msg = {handoff_data, MsgData},
      case gen_fsm:sync_send_all_state_event(VNode, Msg, 60000) of
          ok ->
              State#state{count=Count+1};
          E={error, _} ->
              exit(E)
      end;
  
negative.code:                 peer :: term(),
                  ssl_opts :: [] | list(),
                  tcp_mod :: atom(),
                  recv_timeout_len :: non_neg_integer(),
                  vnode_timeout_len :: non_neg_integer(),
                  partition :: non_neg_integer(),
                  vnode_mod = riak_kv_vnode:: module(),
                  vnode :: pid(),
                  count = 0 :: non_neg_integer()}).
  
  %% set the TCP receive timeout to five minutes to be conservative.
  -define(RECV_TIMEOUT, 300000).
  %% set the timeout for the vnode to process the handoff_data msg to 60s
  -define(VNODE_TIMEOUT, 60000).
  
  start_link() ->
      start_link([]).
  