anchor.diff:           validate_arg/1]).
   -export([reduce_compat/1]).
   -export([no_input_run_reduce_once/0]).
  +
  +-ifdef(TEST).
  +-include_lib("eunit/include/eunit.hrl").
  +-endif.
   
   -include_lib("riak_pipe/include/riak_pipe.hrl").
   -include_lib("riak_pipe/include/riak_pipe_log.hrl").
  
positive.code: %% Mochijson2 conversion will fail on the bare proplist, but will
  %% succeed at encoding this form.
  %%
  %% The exception to the batching controls is handoff.  Whenever a
  %% worker receives handoff from another worker, it immediately reduces
  %% the concatenation of the two inputs.
  %%
  %% If no inputs are received before eoi, this fitting evaluated the
  %% function once, with an empty list as `Inputs'.
  %%
  
negative.code:         false ->
              []
      end.
  
  -ifdef(TEST).
  
  %% This test should check that the reduce function is not called more
  %% often than reduce_phase_batch_size or reduce_phase_only_1 request.
  batch_size_during_handoff_test() ->
      Fun = fun riak_kv_mapreduce:reduce_count_inputs/2,
  
      ReduceEvery5 = [{reduce_phase_batch_size, 5}],
      AInputs = [a,b,c],
      {ok, StateUnreduced} =
          handoff_test_helper(Fun, ReduceEvery5, AInputs, []),
      %% handing off three unprocessed inputs to a fresh worker with
      %% zero unprocessed inputs should not immediately process the
      %% inputs if the batch size is greater than 3
      ?assertEqual(length(AInputs), StateUnreduced#state.delay),
      ?assertEqual(lists:reverse(AInputs), StateUnreduced#state.inacc),
  
      {ok, StateFinally} =
          test_helper({ok, StateUnreduced}, [e,f]),
      %% just two more inputs should still trigger the reduce
      ?assertEqual(0, StateFinally#state.delay),
      ?assertEqual([], StateFinally#state.inacc),
      
      BInputs = [e,f,g],
      {ok, StateReduced} =
          handoff_test_helper(Fun, ReduceEvery5, AInputs, BInputs),
      %% handing off three unprocessed inputs to a worker that has three
      %% more unprocessed inputs should immediately process the inputs
      %% if the batch size is less than or equal to 6
      ?assertEqual(0, StateReduced#state.delay),
      ?assertEqual([], StateReduced#state.inacc),
      
      ReduceOnce = [reduce_phase_only_1],
      {ok, StateNever} =
          handoff_test_helper(Fun, ReduceOnce, AInputs, BInputs),
      %% handing off unprocessed inputs when reduce_phase_only_1 is set
      %% should never immediately reduce them
      ?assertEqual(length(AInputs++BInputs), StateNever#state.delay),
      ?assertEqual(lists:reverse(AInputs++BInputs), StateNever#state.inacc).
  
  %% Start reducer A and reducer B.  Feed AInputs to A and BInputs to B,
  %% then archive A and handoff its data to B.  Returns B's resulting
  %% state.
  handoff_test_helper(Fun, Arg, AInputs, BInputs) ->
      {ok, StateA} = test_helper(Fun, Arg, AInputs),
      {ok, StateB} = test_helper(Fun, Arg, BInputs),
      {ok, Archive} = archive(StateA),
      handoff(Archive, StateB).
  
  %% Initialize a reducer with the given fun and arg, then pass it the
  %% list of inputs.
  test_helper(Fun, Arg, Inputs) ->
      Fitting = #fitting{pid=self(),
                         ref=make_ref(),
                         chashfun=fun() -> <<0:160/integer>> end,
                         nval=1},
      Details = #fitting_details{fitting=Fitting,
                                 name=batch_size_during_handoff_test,
                                 module=?MODULE,
                                 arg={rct, Fun, Arg},
                                 output=Fitting,
                                 options=[],
                                 q_limit=64},
      test_helper(init(0, Details), Inputs).
  
  %% Pass the list of inputs to the given reducer.
  test_helper({ok, State}, Inputs) ->
      lists:foldl(
        fun(I, {ok, S}) -> process(I, true, S) end,
        {ok, State},
        Inputs).
  
  -endif.
  