anchor.diff:  -include("../ibrowse/ibrowse.hrl").
   
   -record (remote, {
  -    conn = nil,
  -    reqid = nil,
  +    conn,
  +    reqid,
  +    last_seq,
       complete = false,
       count = 0,
  +    partial_chunk = nil,
       reply_to = nil,
       rows = queue:new()
   }).
  
positive.code:         [Code,Hdrs]),
      {stop, {error, list_to_integer(Code)}, State};
  
  handle_info({ibrowse_async_response, Id, Msg}, #remote{reqid=Id} = State) ->
      ?LOG_DEBUG("~p reqid ~p ibrowse_async_response ~p", [?MODULE, Id, Msg]),
      #remote{
          complete = Complete,
          count = Count,
          rows = Rows
      } = State,
      try
          Row = decode_row(Msg),
          case State of
          #remote{reply_to=nil} ->
              {noreply, State#remote{count=Count+1, rows = queue:in(Row, Rows)}};
          #remote{count=0, reply_to=From}->
              gen_server:reply(From, Row),
              {noreply, State#remote{reply_to=nil}}
          end
      catch
      throw:{invalid_json, Msg} ->
          ?LOG_DEBUG("got invalid_json ~p", [Msg]),
          ok = maybe_stream_next(Complete, Count, Id),
          {noreply, State}
      end;
  
  handle_info({ibrowse_async_response_end, Id}, #remote{reqid=Id} = State) ->
      ?LOG_DEBUG("got ibrowse_async_response_end ~p", [State#remote.reply_to]),
  
negative.code: 
  %internal funs
  
  process_response(<<"{\"results\":[\n">>, Id, State) ->
      #remote{
          complete = Complete,
          count = Count
      } = State,
      ok = maybe_stream_next(Complete, Count, Id),
      State;
  process_response(<<"\n],\n\"last_seq\":", LastSeqStr/binary>>, _, State) ->
      LastSeq = list_to_integer(?b2l(hd(re:split(LastSeqStr, "}")))),
      State#remote{last_seq = LastSeq};
  process_response(Chunk, Id, #remote{partial_chunk=nil} = State) ->
      #remote{
          complete = Complete,
          count = Count,
          rows = Rows
      } = State,
      try
          Row = decode_row(Chunk),
          ok = maybe_stream_next(Complete, Count+1, Id),
          case State of
          #remote{reply_to=nil} ->
              State#remote{count=Count+1, rows = queue:in(Row, Rows)};
          #remote{count=0, reply_to=From}->
              gen_server:reply(From, Row),
              State#remote{reply_to=nil}
          end
      catch
      throw:{invalid_json, Bad} ->
          ?LOG_DEBUG("got invalid_json ~p", [Bad]),
          ok = maybe_stream_next(Complete, Count, Id),
          State#remote{partial_chunk = Bad}
      end;
  process_response(Chunk, Id, State) ->
      #remote{
          complete = Complete,
          count = Count,
          partial_chunk = Partial,
          rows = Rows
      } = State,
      try
          Row = decode_row(<<Partial/binary, Chunk/binary>>),
          ok = maybe_stream_next(Complete, Count+1, Id),
          case State of
          #remote{reply_to=nil} ->
              State#remote{count=Count+1, partial_chunk=nil, rows=queue:in(Row,Rows)};
          #remote{count=0, reply_to=From}->
              gen_server:reply(From, Row),
              State#remote{reply_to=nil, partial_chunk=nil}
          end
      catch
      throw:{invalid_json, Bad} ->
          ?LOG_DEBUG("got invalid_json ~p", [Bad]),
          ok = maybe_stream_next(Complete, Count, Id),
          State#remote{partial_chunk = Bad}
      end.
      
  decode_row(<<",\n", Rest/binary>>) ->
      decode_row(Rest);
  decode_row(Row) ->
      ?JSON_DECODE(Row).
  