anchor.diff:                  [StreamRef, OutputPid, OutputRef, Msg]),
               OutputPid ! []
       end,
  -ok.
  -
  +    ok.
  +
  +receive_catalog_query_results(StreamRef, OutputPid, OutputRef) ->
  +    receive
  +        {catalog_query, _ReqId, "$end_of_results", _, _, _, _} ->
  +            io:format("catalog_query_results -> $end_of_results~n", []),
  +            OutputPid ! {catalog_query_response, done, OutputRef};
  +        {catalog_query, StreamRef, Partition, Index, 
  +                        Field, Term, JSONProps} ->
  +            OutputPid ! {catalog_query_response, 
  +                            {Partition, Index, Field, Term, JSONProps},
  +                            OutputRef},
  +            receive_catalog_query_results(StreamRef, OutputPid, OutputRef);
  +        Msg ->
  +            io:format("receive_catalog_query_results: unknown message: ~p~n", [Msg]),
  +            receive_catalog_query_results(StreamRef, OutputPid, OutputRef)
  +    end,
  +    ok.
   
   %% @spec get(state(), BKey :: riak_object:bkey()) ->
   %%         {ok, Val :: binary()} | {error, Reason :: term()}
  
positive.code: -author("John Muellerleile <johnm@basho.com>").
  -export([start/2,stop/1,get/2,put/3,list/1,list_bucket/2,delete/2]).
  -export([fold/3, drop/1, is_empty/1]).
  -export([test_stream/4, test_catalog_query/1]).
  
  -include_lib("eunit/include/eunit.hrl").
  -include("riak_search.hrl").
  
negative.code:         Msg ->
              io:format("receive_info_range_results(~p, ~p, ~p) -> ~p~n",
                  [StreamRef, OutputPid, OutputRef, Msg]),
              receive_info_range_results(StreamRef, OutputPid, OutputRef, Results)
      end,
      ok.
  
  receive_info_results(StreamRef, OutputPid, OutputRef) ->
      receive
  