anchor.diff:  -endif.
   -include_lib("riakc/include/riakclient_pb.hrl").
   -include_lib("riakc/include/riakc_pb.hrl").
  +-include_lib("riak_pipe/include/riak_pipe.hrl").
   -behaviour(gen_server).
   
   -export([start_link/0, set_socket/2]).
  
positive.code: %% TODO: refactor, cleanup
  %% Start map/reduce job - results will be processed in handle_info
  process_message(#rpbmapredreq{request=MrReq, content_type=ContentType}=Req, 
                  #state{client=C} = State) ->
  
      case decode_mapred_query(MrReq, ContentType) of
          {error, Reason} ->
              send_error("~p", [Reason], State);
  
          {ok, Inputs, Query, Timeout} ->
              case is_binary(Inputs) orelse is_key_filter(Inputs) of
                  true ->
                      case C:mapred_bucket_stream(Inputs, Query, 
                                                  self(), Timeout) of
                          {stop, Error} ->
                              send_error("~p", [Error], State);
  
                          {ok, ReqId} ->
                              {pause, State#state{req = Req, req_ctx = ReqId}}
                      end;
                  false ->
                      case is_list(Inputs) of
                          true ->
                              case C:mapred_stream(Query, self(), Timeout) of
                                  {stop, Error} ->
                                      send_error("~p", [Error], State);
                                  
                                  {ok, {ReqId, FSM}} ->
                                      luke_flow:add_inputs(FSM, Inputs),
                                      luke_flow:finish_inputs(FSM),
                                      %% Pause incoming packets - map/reduce results
                                      %% will be processed by handle_info, it will 
                                      %% set socket active again on completion of streaming.
                                      {pause, State#state{req = Req, req_ctx = ReqId}}
                              end;
                          false ->
                              case is_tuple(Inputs) andalso size(Inputs)==4 andalso
                                  element(1, Inputs) == modfun andalso
                                  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);
                  
                                          {ok, {ReqId, FSM}} ->
                                              C:mapred_dynamic_inputs_stream(
                                                FSM, Inputs, Timeout),
                                              luke_flow:finish_inputs(FSM),
                                              %% Pause incoming packets - map/reduce results
                                              %% will be processed by handle_info, it will 
                                              %% set socket active again on completion of streaming.
                                              {pause, State#state{req = Req, req_ctx = ReqId}}
                                      end;
                                  false -> 
                                      {error, bad_mapred_inputs}
                              end
                          end
                      end
              end.
  
  %% Send a message to the client
  -spec send_msg(msg(), #state{}) -> #state{}.
  
negative.code: handle_info({ReqId, {keys, Keys}}, State=#state{req=#rpblistkeysreq{}, req_ctx=ReqId}) ->
      {noreply, send_msg(#rpblistkeysresp{keys = Keys}, State)};
  
  %% PIPE Handle response from mapred_stream
  handle_info(#pipe_eoi{ref=ReqId},
              State=#state{sock = Socket, req=#rpbmapredreq{}, req_ctx=ReqId}) ->
      NewState = send_msg(#rpbmapredresp{done = 1}, State),
      inet:setopts(Socket, [{active, once}]),
      {noreply, NewState#state{req = undefined, req_ctx = undefined}};
  
  handle_info(#pipe_result{ref=ReqId, from=PhaseId, result=Res},
              State=#state{sock=Socket,
                           req=#rpbmapredreq{content_type = ContentType}, 
                           req_ctx=ReqId}) ->
      case encode_mapred_phase([Res], ContentType) of
          {error, Reason} ->
              NewState = send_error("~p", [Reason], State),
              inet:setopts(Socket, [{active, once}]),
              {noreply, NewState#state{req = undefined, req_ctx = undefined}};
          Response ->
              {noreply, send_msg(#rpbmapredresp{phase=PhaseId, 
                                                response=Response}, State)}
      end;
  %% ignore #pipe_log for now, since riak_kv_mrc_pipe does not enable it
  
  %% LEGACY Handle response from mapred_stream/mapred_bucket_stream
  handle_info({flow_results, ReqId, done},
              State=#state{sock = Socket, req=#rpbmapredreq{}, req_ctx=ReqId}) ->
      NewState = send_msg(#rpbmapredresp{done = 1}, State),
  