anchor.diff:              case is_binary(Inputs) orelse is_key_filter(Inputs) of
                   true ->
                       case C:mapred_bucket_stream(Inputs, Query, 
  -                                                self(), Timeout) of
  +                                                self(), ResultTransformer, Timeout) of
                           {stop, Error} ->
                               send_error("~p", [Error], State);
   
  
positive.code:                                 is_atom(element(2, Inputs)) andalso
                                  is_atom(element(3, Inputs)) of
                                  true ->
                                      case C:mapred_stream(Query, self(), Timeout) of
                                          {stop, Error} ->
                                              send_error("~p", [Error], State);
                  
  
negative.code: normalize_rw_value(?RIAKC_RW_ALL) -> all;
  normalize_rw_value(?RIAKC_RW_DEFAULT) -> default;
  normalize_rw_value(V) -> V.
  
  %% get a result transformer for the content-type
  %% jsonify not_founds for application/json
  %% do nothing otherwise
  get_result_transformer(<<"application/json">>) ->
      fun riak_kv_mapred_json:jsonify_not_found/1;
  get_result_transformer(_) ->
      fun(X) ->
              X end.
  