anchor.diff:                      {stop, Acc2} = Callback({stream_end,
                                                SnapshotStart, SnapshotEnd, LastSeenSeqno}, Acc),
                       consumer_loop_exit(Child, done, RestData),
  -                    Acc2;
  -                #dcp_packet{opcode = ?DCP_NOP} ->
  -                    consume_stuff_loop(Child, Callback, Acc, ConsumedSoFar,
  -                                       SnapshotStart, SnapshotEnd, LastSeenSeqno,
  -                                       RestData)
  +                    Acc2
               end;
           Data ->
               consumer_loop_recv(Child, Callback, Acc, ConsumedSoFar,
  
positive.code:             self() ! {tcp, Socket, Data}
      end,
      inet:setopts(Socket, [{active, true}]),
      socket_loop(Socket, Callback, Acc, <<>>, Consumer).
  
  socket_loop(Socket, Callback, Acc, Data, Consumer) ->
      Msg = receive
                XMsg ->
                    XMsg
            end,
      case Msg of
          {tcp, _Socket, NewData0} ->
              NewData = case Data of
                            <<>> ->
                                NewData0;
                            _ ->
                                <<Data/binary, NewData0/binary>>
                        end,
              SplitPos = nops_loop(Socket, NewData, 0),
              <<ScannedData:SplitPos/binary, UnscannedData/binary>> = NewData,
              case SplitPos =/= 0 of
                  true ->
                      Consumer ! ScannedData;
                  _ ->
                      ok
              end,
              socket_loop(Socket, Callback, Acc, UnscannedData, Consumer);
          {tcp_closed, MustSocket} ->
              {tcp_closed_socket, Socket} = {tcp_closed_socket, MustSocket},
              erlang:error(premature_socket_closure);
          ConsumedBytes when is_integer(ConsumedBytes) ->
              ok = gen_tcp:send(Socket, encode_req(#dcp_packet{opcode = ?DCP_WINDOW_UPDATE,
                                                               ext = <<ConsumedBytes:32/big>>})),
              socket_loop(Socket, Callback, Acc, Data, Consumer);
          done ->
              ok = gen_tcp:send(Socket, encode_req(#dcp_packet{opcode = ?DCP_WINDOW_UPDATE,
                                                               ext = <<(?XDCR_DCP_BUFFER_SIZE):32/big>>}));
  
negative.code:             ?log_debug("Got stop. Dropping socket on the floor")
      end.
  
  socket_loop_enter(Socket, Callback, Acc, Data, Consumer) ->
      case Data of
          <<>> ->
  