anchor.diff:                  false ->
                       case is_list(Inputs) of
                           true ->
  -                            case C:mapred_stream(Query, self(), Timeout) of
  +                            case C:mapred_stream(Query, 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:             case is_binary(Inputs) orelse is_key_filter(Inputs) of
                  true ->
                      case C:mapred_bucket_stream(Inputs, Query, 
                                                  self(), ResultTransformer, Timeout) of
                          {stop, Error} ->
                              send_error("~p", [Error], State);
  
  