anchor.diff:                      Acc2 = Acc,
                       skip
               end,
  -            receive_stream_results(StreamRef, Sender, FilterFun, Acc2);
  -        _Msg ->
  -            %% TODO: Should this throw - must be an error
  -            %% io:format("receive_stream_results(~p, ~p, ~p) -> ~p~n",
  -            %%     [StreamRef, Sender, FilterFun, _Msg]),
  -            riak_search_backend:stream_response_done(Sender)
  +            receive_stream_results(StreamRef, Sender, FilterFun, Conn, Acc2);
  +        Msg ->
  +            riak_search_backend:stream_response_done(Sender),
  +            raptor_conn:close(Conn),
  +            throw({unexpected_msg, Msg})
  +    after
  +        ?STREAM_TIMEOUT ->
  +            riak_search_backend:stream_response_done(Sender),
  +            raptor_conn:close(Conn),
  +            error_logger:warning_msg("Stream result Raptor conn timeout\n")
       end,
       ok.
   
  -receive_info_range_results(StreamRef, Sender) ->
  -    receive_info_range_results(StreamRef, Sender, []).
  -
  -receive_info_range_results(StreamRef, Sender, Results) ->
  +receive_info_range_results(StreamRef, Sender, Conn) ->
  +    receive_info_range_results(StreamRef, Sender, Conn, []).
  +
  +receive_info_range_results(StreamRef, Sender, Conn, Results) ->
       receive
           {info, StreamRef, timeout} ->
  -            riak_search_backend:info_response(Sender, Results);
  +            riak_search_backend:info_response(Sender, Results),
  +            raptor_conn:close(Conn),
  +            error_logger:warning_msg("Info range result Raptor socket timeout\n");
  +
           {info, StreamRef, "$end_of_info", 0} ->
               riak_search_backend:info_response(Sender, Results);
           
           %% TODO: Replace this with a [New | Acc] and lists:reverse
           {info, StreamRef, Term, Count} ->
  -            receive_info_range_results(StreamRef, Sender,
  -                Results ++ [{Term, node(), Count}])
  -
  +            receive_info_range_results(StreamRef, Sender, Conn,
  +                Results ++ [{Term, node(), Count}]);
  +        Msg ->
  +            riak_search_backend:info_response(Sender, [Results]),
  +            raptor_conn:close(Conn),
  +            throw({unexpected_msg, Msg})
  +    after
  +        ?INFO_RANGE_TIMEOUT ->
  +            riak_search_backend:info_response(Sender, Results),
  +            raptor_conn:close(Conn),
  +            error_logger:warning_msg("Info range result Raptor conn timeout\n")
       end,
       ok.
   
  -receive_info_results(StreamRef, Sender) ->
  +receive_info_results(StreamRef, Sender, Conn) ->
       receive
           {info, StreamRef, timeout} ->
  +            riak_search_backend:info_response(Sender, []),
  +            raptor_conn:close(Conn),
  +            error_logger:warning_msg("Info result Raptor socket timeout\n"),
               ok;
           {info, StreamRef, "$end_of_info", _Count} ->
               ok;
           {info, StreamRef, Term, Count} ->
               riak_search_backend:info_response(Sender, [{Term, node(), Count}]),
  -            receive_info_results(StreamRef, Sender)
  -    after
  -        ?INFO_TIMEOUT ->
  -            ok
  +            receive_info_results(StreamRef, Sender, Conn);
  +        Msg ->
  +            riak_search_backend:info_response(Sender, []),
  +            raptor_conn:close(Conn),
  +            throw({unexpected_msg, Msg})
  +    after
  +        ?INFO_TIMEOUT ->           
  +            riak_search_backend:info_response(Sender, []),
  +            raptor_conn:close(Conn),
  +            error_logger:warning_msg("Info result Raptor conn timeout\n")
       end,
       ok.
   
  -receive_catalog_query_results(StreamRef, Sender) ->
  +receive_catalog_query_results(StreamRef, Sender, Conn) ->
       receive
           {catalog_query, _ReqId, timeout} ->
  -            riak_search_backend:catalog_query_done(Sender);
  +            riak_search_backend:catalog_query_done(Sender),
  +            raptor_conn:close(Conn),
  +            error_logger:warning_msg("Catalog query result Raptor socket timeout\n");
           {catalog_query, _ReqId, "$end_of_results", _, _, _, _} ->
               riak_search_backend:catalog_query_done(Sender);
           {catalog_query, StreamRef, Partition, Index,
  
positive.code:                                                to_binary(Field),
                                                 to_binary(Term),
                                                 Partition),
                             receive_info_results(StreamRef, Sender)
                         after
                             riak_sock_pool:checkin(?CONN_POOL, Conn)
                         end
  
negative.code:             Val = {Value, Props2},
              FoldResultPid ! {fold_result, Key, Val},
              fold_stream_process(CatalogProcessPid, FoldResultPid, StreamRef, 
                                  IndexBin, FieldBin, TermBin, Conn);
          Msg ->
              FoldResultPid ! {fold_stream, done, StreamRef},
              raptor_conn:close(Conn),
              throw({unexpected_msg, Msg})
      after ?FOLD_TIMEOUT ->
              CatalogProcessPid ! {fold_stream, done, StreamRef},
              raptor_conn:close(Conn),
              error_logger:warning_msg("fold_stream_process: raptor conn timed out: ~p.~p.~p~n",
                                        [IndexBin, FieldBin, TermBin])
      end.
  
  