riak_postcommit_hook_example.erl

-module(riak_postcommit_hook_example).
-export([postcommitcounter/1, postcommitmapcounter/1]).

%% Refreshing counters from commithook is very expensive!
%%
%% requirements: bucket-type: maps(Type::map), bucket-type: counters(Type::counters)
%%
%% riak-admin bucket-type create maps '{"props":{"datatype":"map"}}'
%% riak-admin bucket-type activate maps
%% riak-admin bucket-type create counters '{"props":{"datatype":"counter"}}'
%% riak-admin bucket-type activate counters
%%
%% you have to update if you have modified the source:
%%
%% compile the source with same erl version:
%% erlc riak_postcommit_hook_example.erl
%% 
%% copy to riak's erlang(PA) directory:
%% // PA directory is configurable: /etc/advanced.config -> {riak_kv, [{add_paths, ["/opt/riak.2.2.3/erl_commit_hooks_and_utils"]} ]}
%% cp riak_postcommit_hook_example.beam /opt/riak.2.2.3/erl_commit_hooks_and_utils 
%%
%% reload erl beams:
%% riak-admin erl-reload
%% 
%% author: Szloboda Zsolt, [email protected]
%%

postcommitcounter(RObj) ->
    {ok, Client} = riak:local_client(),
    Bucket = {<<"counters">>, <<"test">>},
    Id = <<"all">>,
    Counter = case Client:get(Bucket, Id) of
                  {error, notfound} -> riak_kv_crdt:new(Bucket, Id, riak_dt_pncounter);
                  {ok, Obj} -> Obj
              end,

    Op = {crdt_op, riak_dt_pncounter, {increment, 1}, undefined},
    Counter2 = riak_kv_crdt:update(Counter, mkhash(), Op),

    %%Client:put(Counter2),
    
    %% speedup
    riak_kv_w1c_worker:put(Counter2,[]),
    RObj.


postcommitmapcounter(RObj) ->
    {ok, Client} = riak:local_client(),
    Bucket = {<<"maps">>, <<"test">>},
    Id = <<"all">>,
    {Counter, Context} = case Client:get(Bucket, Id) of
                             {error, notfound} ->
                                 Obj = riak_kv_crdt:new(Bucket, Id, riak_dt_map),
                                 {Obj, undefined};
                             {ok, Obj} ->
                                 {{CountContext,_},_} = riak_kv_crdt:value(Obj, riak_dt_map),
                                 {Obj, CountContext}
                         end,

    Op = {crdt_op, riak_dt_map, {update,[{update,{<<"count">>,riak_dt_emcntr},increment}]}, Context},
    Counter2 = riak_kv_crdt:update(Counter, mkhash(), Op),

    %%Client:put(Counter2),
    
    %% speedup
    riak_kv_w1c_worker:put(Counter2,[]),
    RObj.

mkhash() ->
    integer_to_binary(erlang:phash2({self(), os:timestamp()})).