anchor.diff:  %% ===================================================================
   %% Internal functions
   %% ===================================================================
  +
  +%% @private get a Data Type value
  +fetch_type({ok, {true, Type}}, Req, State) ->
  +    Mod = riak_kv_crdt:to_mod(Type),
  +    Supported = riak_kv_crdt:supported(Mod),
  +    maybe_fetch(Supported, Req, State#state{mod=Mod, type=Type});
  +fetch_type({ok, {false, _Type}}, _Req, State) ->
  +    {error, {format, "Bucket must be allow_mult=true"}, State};
  +fetch_type({error, no_type}, #dtfetchreq{type=BType}, State) ->
  +    {error, {format, "Error no bucket type `~p`", [BType]}, State}.
  +
  +maybe_fetch(true, Req, State) ->
  +    #dtfetchreq{bucket=B, key=K, type=BType, include_context=InclCtx} = Req,
  +    #state{client=C} = State,
  +    Options = make_options(Req),
  +    Resp = C:get({BType, B}, K, Options),
  +    process_fetch_response(Resp, State#state{return_ctx=InclCtx});
  +maybe_fetch(false, _Req, State) ->
  +    #state{type=Type} = State,
  +    {error, {format, "`~p` is not a supported type", [Type]}, State}.
  +
  +%% @private prepare a repsonse for a fetch operation
  +process_fetch_response({ok, O}, State) ->
  +    #state{type=Type, mod=Mod, return_ctx=ReturnCtx} = State,
  +    {Ctx0, Value} = riak_kv_crdt:value(O, Mod),
  +    Ctx = get_context(Ctx0, ReturnCtx),
  +    Resp = riak_pb_dt_codec:encode_fetch_response(Type, Value, Ctx, ?MOD_MAP),
  +    {reply, Resp, State};
  +process_fetch_response({error, notfound}, State) ->
  +    #state{type=Type} = State,
  +    Resp = riak_pb_dt_codec:encode_fetch_response(Type, undefined, undefined),
  +    {reply, Resp, State};
  +process_fetch_response({error, Reason}, State) ->
  +    {error, {format,Reason}, State}.
  +
  +update_type({ok, {true, Type}}, #dtupdatereq{op=Op0}=Req, State0) ->
  +    Mod = riak_kv_crdt:to_mod(Type),
  +    Supported = riak_kv_crdt:supported(Mod),
  +    Op = riak_pb_dt_codec:decode_operation(Op0, ?MOD_MAP),
  +    OpType = riak_pb_dt_codec:operation_type(Op0),
  +    ModsMatch = mods_match(Mod, OpType),
  +    State = State0#state{mod=Mod, type=Type, op=Op, op_type=OpType},
  +    maybe_update({Supported, ModsMatch}, Req, State);
  +update_type({ok, {false, _Type}}, _Req, State) ->
  +    {error, {format, "Bucket must be allow_mult=true"}, State};
  +update_type({error, no_type}, #dtupdatereq{type=BType}, State) ->
  +    {error, {format, "Error no bucket type `~p`", [BType]}, State}.
  +
  +maybe_update({true, true}, Req, State0) ->
  +    #dtupdatereq{bucket=B, key=K, type=BType,
  +                 include_context=InclCtx,
  +                 context=Ctx} = Req,
  +    #state{client=C, mod=Mod, op=Op} = State0,
  +    {Key, ReturnKey} = get_key(K),
  +    O = riak_kv_crdt:new({BType, B}, Key, Mod),
  +    Options0 = make_options(Req),
  +    CrdtOp = make_operation(Mod, Op, Ctx),
  +    Options = [{crdt_op, CrdtOp}]  ++ Options0,
  +    Resp =  C:put(O, Options),
  +    State = State0#state{return_key=ReturnKey, return_ctx=InclCtx},
  +    process_update_response(Resp, State);
  +maybe_update({false, _}, _Req, State ) ->
  +    #state{type=Type} = State,
  +    {error, {format, "Bucket datatype `~p` is not a supported type", [Type]}, State};
  +maybe_update({_, false}, _Req, State) ->
  +    #state{type=Type, op_type=OpType} = State,
  +    {error, {format, "Operation type is `~p` but  bucket type is `~p`.",
  +             [OpType, Type]}, State}.
  +
  +process_update_response(ok, State) ->
  +    #state{return_key=ReturnKey} = State,
  +    {reply, #dtupdateresp{key=ReturnKey}, State};
  +process_update_response({ok, RObj}, State) ->
  +    #state{type=Type, return_ctx=ReturnCtx} = State,
  +    {Ctx, Value} = riak_kv_crdt:value(RObj, Type),
  +    {reply, update_resp(Type, Value, get_context(Ctx, ReturnCtx)), State};
  +process_update_response({error, notfound}, State) ->
  +    {reply, #dtupdateresp{}, State};
  +process_update_response({error, Reason}, State) ->
  +    {error, {format, Reason}, State}.
  +
  +%% Make a list of options for the Get / Put ops
  +make_options(#dtfetchreq{r=R0, pr=PR0,
  +                         notfound_ok=NFOk, basic_quorum=BQ,
  +                         sloppy_quorum=SloppyQ, n_val=NVal}) ->
  +    R = decode_quorum(R0),
  +    PR = decode_quorum(PR0),
  +    make_option(r, R) ++
  +        make_option(pr, PR) ++
  +        make_option(notfound_ok, NFOk) ++
  +        make_option(basic_quorum, BQ) ++
  +        make_option(sloppy_quorum, SloppyQ) ++
  +        make_option(n_val, NVal);
  +make_options(#dtupdatereq{w=W0, dw=DW0, pw=PW0,
  +                          timeout=Timeout, sloppy_quorum=SloppyQ,
  +                          n_val=NVal, return_body=RetVal}) ->
  +    W = decode_quorum(W0),
  +    DW = decode_quorum(DW0),
  +    PW = decode_quorum(PW0),
  +    make_option(w, W) ++
  +        make_option(dw, DW) ++
  +        make_option(pw, PW) ++
  +        make_option(timeout, timeout(Timeout)) ++
  +        make_option(sloppy_quorum, SloppyQ) ++
  +        make_option(n_val, NVal) ++ return_value(RetVal).
   
   %% return a key/value tuple that we can ++ to other options so long as the
   %% value is not default or undefined -- those values are pulled from the
  
positive.code: 
  -import(riak_pb_kv_codec, [decode_quorum/1]).
  
  -record(state, {client}).
  
  -define(DEFAULT_TIMEOUT, 60000).
  
  %% @doc init/0 callback. Returns the service internal start
  %% state.
  
negative.code:     {ok, riak_pb_codec:encode(Message)}.
  
  %% @doc process/2 callback. Handles an incoming request message.
  process(#dtfetchreq{type=?DEFAULT_BUCKET}=Req0, State) ->
      %% Handle a typeless message
      %% @see downgrade_request/1 etc below for details
      Req = downgrade_request(Req0),
      process_legacy_counter(Req, State);
  process(#dtfetchreq{bucket=B, type=BType}=Req, State) ->
      %% V2 fetch operation
      fetch_type(bucket_type_to_type(B, BType), Req, State);
  process(#dtupdatereq{type=?DEFAULT_BUCKET}=Req0, State) ->
      %% Handle a typeless update message
      %% @see downgrade_request/1 etc below for details
      Req = downgrade_request(Req0),
      process_legacy_counter(Req, State);
  process(#dtupdatereq{bucket=B, type=BType}=Req, State) ->
      %% V2 update operation
      update_type(bucket_type_to_type(B, BType), Req, State).
  
  %% @doc process_stream/3 callback. This service does not create any
  %% streaming responses and so ignores all incoming messages.
  