anchor.diff:              self() ! {tcp, Socket, Data}
       end,
       inet:setopts(Socket, [{active, true}]),
  -    socket_loop(Socket, Callback, Acc, <<>>, Consumer).
  -
  -socket_loop(Socket, Callback, Acc, Data, Consumer) ->
  +    socket_loop(Socket, Callback, Acc, Consumer).
  +
  +socket_loop(Socket, Callback, Acc, 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, _Socket, NewData} ->
  +            Consumer ! NewData,
  +            socket_loop(Socket, Callback, Acc, 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);
  +            socket_loop(Socket, Callback, Acc, Consumer);
           done ->
               ok = gen_tcp:send(Socket, encode_req(#dcp_packet{opcode = ?DCP_WINDOW_UPDATE,
                                                                ext = <<(?XDCR_DCP_BUFFER_SIZE):32/big>>}));
  
positive.code:                     {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)
              end;
          Data ->
              consumer_loop_recv(Child, Callback, Acc, ConsumedSoFar,
  
negative.code:             ?log_debug("Got stop. Dropping socket on the floor")
      end.
  
  socket_loop_enter(Socket, Callback, Acc, Data, Consumer) ->
      case Data of
          <<>> ->
  