anchor.diff:  
   consume_stuff_loop(Child, Callback, Acc, ConsumedSoFar,
                      SnapshotStart, SnapshotEnd, LastSeenSeqno,
  -                   Data) ->
  -    case decode_packet(Data) of
  -        {_Type, Packet, RestData, PacketSize} ->
  -            case Packet of
  -                #dcp_packet{opcode = ?DCP_MUTATION,
  -                            datatype = DT,
  -                            cas = CAS,
  -                            ext = Ext,
  -                            key = Key,
  -                            body = Body} ->
  -                    <<Seq:64, RevSeqno:64, Flags:32, Expiration:32, _/binary>> = Ext,
  -                    Rev = {RevSeqno, <<CAS:64, Expiration:32, Flags:32>>},
  -                    Doc = #dcp_mutation{id = Key,
  -                                        local_seq = Seq,
  -                                        rev = Rev,
  -                                        body = Body,
  -                                        datatype = DT,
  -                                        deleted = false,
  -                                        snapshot_start_seq = SnapshotStart,
  -                                        snapshot_end_seq = SnapshotEnd},
  -                    consume_stuff_call_callback(Doc,
  -                                                Child, Callback, Acc, ConsumedSoFar + PacketSize,
  -                                                SnapshotStart, SnapshotEnd, Seq,
  -                                                RestData);
  -                #dcp_packet{opcode = ?DCP_SNAPSHOT_MARKER, ext = Ext} ->
  -                    <<NewSnapshotStart:64, NewSnapshotEnd:64, _/binary>> = Ext,
  -                    consume_stuff_loop(Child, Callback, Acc, ConsumedSoFar + PacketSize,
  -                                       NewSnapshotStart, NewSnapshotEnd, LastSeenSeqno,
  -                                       RestData);
  -                #dcp_packet{opcode = ?DCP_DELETION,
  -                            cas = CAS,
  -                            ext = Ext,
  -                            key = Key} ->
  -                    <<Seq:64, RevSeqno:64, _/binary>> = Ext,
  -                    %% NOTE: as of now dcp doesn't expose flags of deleted
  -                    %% docs
  -                    Rev = {RevSeqno, <<CAS:64, 0:32, 0:32>>},
  -                    Doc = #dcp_mutation{id = Key,
  -                                        local_seq = Seq,
  -                                        rev = Rev,
  -                                        body = <<>>,
  -                                        datatype = 0,
  -                                        deleted = true,
  -                                        snapshot_start_seq = SnapshotStart,
  -                                        snapshot_end_seq = SnapshotEnd},
  -                    consume_stuff_call_callback(Doc,
  -                                                Child, Callback, Acc, ConsumedSoFar + PacketSize,
  -                                                SnapshotStart, SnapshotEnd, Seq,
  -                                                RestData);
  -                #dcp_packet{opcode = ?DCP_STREAM_END} ->
  -                    {stop, Acc2} = Callback({stream_end,
  -                                             SnapshotStart, SnapshotEnd, LastSeenSeqno}, Acc),
  -                    consumer_loop_exit(Child, done, RestData),
  -                    Acc2
  +                   Data, Len, NeedLen) ->
  +    case Len >= NeedLen of
  +        true ->
  +            case try_decode_packet(Data) of
  +                {ok, {_Type, Packet, RestData, PacketSize}} ->
  +                    do_consume_stuff_loop(Child, Callback, Acc,
  +                                          ConsumedSoFar + PacketSize,
  +                                          SnapshotStart, SnapshotEnd, LastSeenSeqno,
  +                                          RestData, Len - PacketSize, 0,
  +                                          Packet);
  +                {need_more_data, NewNeedLen} ->
  +                    consumer_loop_recv(Child, Callback, Acc, ConsumedSoFar,
  +                                       SnapshotStart, SnapshotEnd, LastSeenSeqno,
  +                                       Data, Len, NewNeedLen)
               end;
  -        Data ->
  +        false ->
               consumer_loop_recv(Child, Callback, Acc, ConsumedSoFar,
                                  SnapshotStart, SnapshotEnd, LastSeenSeqno,
  -                               Data)
  -    end.
  -
  --compile({inline, [consume_stuff_call_callback/9]}).
  +                               Data, Len, NeedLen)
  +    end.
  +
  +do_consume_stuff_loop(Child, Callback, Acc, ConsumedSoFar,
  +                      SnapshotStart, SnapshotEnd, LastSeenSeqno,
  +                      Data, Len, NeedLen, Packet) ->
  +    case Packet of
  +        #dcp_packet{opcode = ?DCP_MUTATION,
  +                    datatype = DT,
  +                    cas = CAS,
  +                    ext = Ext,
  +                    key = Key,
  +                    body = Body} ->
  +            <<Seq:64, RevSeqno:64, Flags:32, Expiration:32, _/binary>> = Ext,
  +            Rev = {RevSeqno, <<CAS:64, Expiration:32, Flags:32>>},
  +            Doc = #dcp_mutation{id = Key,
  +                                local_seq = Seq,
  +                                rev = Rev,
  +                                body = Body,
  +                                datatype = DT,
  +                                deleted = false,
  +                                snapshot_start_seq = SnapshotStart,
  +                                snapshot_end_seq = SnapshotEnd},
  +            consume_stuff_call_callback(Doc,
  +                                        Child, Callback, Acc, ConsumedSoFar,
  +                                        SnapshotStart, SnapshotEnd, Seq,
  +                                        Data, Len, NeedLen);
  +        #dcp_packet{opcode = ?DCP_SNAPSHOT_MARKER, ext = Ext} ->
  +            <<NewSnapshotStart:64, NewSnapshotEnd:64, _/binary>> = Ext,
  +            consume_stuff_loop(Child, Callback, Acc, ConsumedSoFar,
  +                               NewSnapshotStart, NewSnapshotEnd, LastSeenSeqno,
  +                               Data, Len, NeedLen);
  +        #dcp_packet{opcode = ?DCP_DELETION,
  +                    cas = CAS,
  +                    ext = Ext,
  +                    key = Key} ->
  +            <<Seq:64, RevSeqno:64, _/binary>> = Ext,
  +            %% NOTE: as of now dcp doesn't expose flags of deleted
  +            %% docs
  +            Rev = {RevSeqno, <<CAS:64, 0:32, 0:32>>},
  +            Doc = #dcp_mutation{id = Key,
  +                                local_seq = Seq,
  +                                rev = Rev,
  +                                body = <<>>,
  +                                datatype = 0,
  +                                deleted = true,
  +                                snapshot_start_seq = SnapshotStart,
  +                                snapshot_end_seq = SnapshotEnd},
  +            consume_stuff_call_callback(Doc,
  +                                        Child, Callback, Acc, ConsumedSoFar,
  +                                        SnapshotStart, SnapshotEnd, Seq,
  +                                        Data, Len, NeedLen);
  +        #dcp_packet{opcode = ?DCP_STREAM_END} ->
  +            {stop, Acc2} = Callback({stream_end,
  +                                     SnapshotStart, SnapshotEnd, LastSeenSeqno}, Acc),
  +            consumer_loop_exit(Child, done, Data),
  +            Acc2
  +    end.
  +
  +-compile({inline, [consume_stuff_call_callback/11]}).
   
   consume_stuff_call_callback(Doc, Child, Callback, Acc, ConsumedSoFar,
                               SnapshotStart, SnapshotEnd, Seq,
  -                            RestData) ->
  +                            Data, Len, NeedLen) ->
       erlang:put(last_doc, Doc),
       case Callback(Doc, Acc) of
           {ok, Acc2} ->
               consume_stuff_loop(Child, Callback, Acc2, ConsumedSoFar,
                                  SnapshotStart, SnapshotEnd, Seq,
  -                               RestData);
  +                               Data, Len, NeedLen);
           {stop, Acc2} ->
  -            consumer_loop_exit(Child, stop, []),
  +            consumer_loop_exit(Child, stop, <<>>),
               Acc2
       end.
   
  
positive.code: encode_res(Packet) ->
      encode_req(Packet#dcp_packet{vbucket = Packet#dcp_packet.status}).
  
  decode_packet(<<Magic:8, Opcode:8, KeyLen:16,
                  ExtLen:8, DT:8, VB:16,
                  BodyLen:32,
                  Opaque:32,
                  Cas:64, Rest/binary>> = FullBinary) ->
      case Rest of
          <<Body:BodyLen/binary, RestRest/binary>> ->
              <<Ext:ExtLen/binary, KB/binary>> = Body,
              <<Key0:KeyLen/binary, TrueBody0/binary>> = KB,
              Key = binary:copy(Key0),
              TrueBody = binary:copy(TrueBody0),
              case Magic of
                  ?REQ_MAGIC ->
                      {req,
                       #dcp_packet{opcode = Opcode,
                                   datatype = DT,
                                   vbucket = VB,
                                   opaque = Opaque,
                                   cas = Cas,
                                   ext = Ext,
                                   key = Key,
                                   body = TrueBody},
                       RestRest,
                       ?HEADER_LEN + BodyLen};
                  ?RES_MAGIC ->
                      {res,
                       #dcp_packet{opcode = Opcode,
                                   datatype = DT,
                                   status = VB,
                                   opaque = Opaque,
                                   cas = Cas,
                                   ext = Ext,
                                   key = Key,
                                   body = TrueBody},
                       RestRest,
                       ?HEADER_LEN + BodyLen}
              end;
          _ ->
              FullBinary
      end;
  decode_packet(FullBinary) ->
      FullBinary.
  
  build_stream_request_packet(Vb, Opaque,
                              StartSeqno, EndSeqno, VBUUID,
  
negative.code:         {failover_id, _FailoverUUID, StartSeqno, _, SnapshotStart, SnapshotEnd} = Evt ->
              {ok, Acc2} = Callback(Evt, Acc),
              consumer_loop_recv(Child, Callback, Acc2, 0,
                                 SnapshotStart, SnapshotEnd, StartSeqno,
                                 <<>>, 0, 0)
      end.
  
  consumer_loop_recv(Child, Callback, Acc, ConsumedSoFar0,
                     SnapshotStart, SnapshotEnd, LastSeenSeqno,
                     Data, Len, NeedLen) ->
      ConsumedSoFar =
          case ConsumedSoFar0 >= ?XDCR_DCP_BUFFER_SIZE div 3 of
              true ->
  