anchor.diff:      ObjectList = term_to_binary(Items),
   
       M = <<?PT_MSG_BATCH:8, ObjectList/binary>>,
  -    
  +
       NumBytes = byte_size(M),
  -    
  +
       Stats2 = incr_bytes(incr_objs(Stats, NObjects), NumBytes),
       Stats3 = maybe_send_status({Module, SrcPartition, TargetPartition}, Stats2),
  -    
  +
       case TcpMod:send(Sock, M) of
           ok ->
  -            Acc#ho_acc{ack=Ack+1, error=ok, stats=Stats3, 
  +            Acc#ho_acc{ack=Ack+1, error=ok, stats=Stats3,
                          total_objects=TotalObjects+NObjects,
                          total_bytes=TotalBytes+NumBytes,
                          item_queue=[],
  
positive.code:         Filter -> Filter
      end.
  
  %% @private 
  %%
  %% @doc check if the handoff reciever will accept batching messages
  %%      otherwise fall back to the slower, object-at-a-time path
  
  remote_supports_batching(Node) ->
  
      case catch rpc:call(Node, riak_core_handoff_receiver, 
                    supports_batching, []) of
          true ->
              lager:debug("remote node supports batching, enabling"),
  
negative.code:            total_objects=TotalObjects,
             total_bytes=TotalBytes,
             stats=FinalStats,
             acksync_timer=TRef,
             notsent_acc=NotSentAcc} = AccRecord,
  
           _ = timer:cancel(TRef),
           case ErrStatus of
               ok ->
                   %% One last sync to make sure the message has been received.
  