Merge pull request #1081 from basho/feature/zl/helper-fn-check-if-obj…

…ect-bucket-props-supports-crdt

add is_crdt convenience fn

Reviewed-by: seancribbs

src/riak_kv_crdt.erl

@@ -28,6 +28,8 @@
 -export([log_merge_errors/4, meta/2, merge_value/2]).
 %% MR helper funs
 -export([value/1, counter_value/1, set_value/1, map_value/1]).
+%% Other helper funs
+-export([is_crdt/1]).
 
 -include("riak_kv_wm_raw.hrl").
 -include("riak_object.hrl").
@@ -132,6 +134,21 @@ map_value(RObj) ->
     {{_Ctx, Map}, _Stats}  = value(RObj, ?MAP_TYPE),
     Map.
 
+%% @doc convenience function for (e.g.) Yokozuna. Checks the bucket props for
+%% the object, if it has a supported datatype entry, returns true; otherwise
+%% false if not a 2.0 CRDT.
+-spec is_crdt(riak_object:riak_object()) -> boolean()|{error,_}.
+is_crdt(RObj) ->
+    Bucket = riak_object:bucket(RObj),
+    case riak_core_bucket:get_bucket(Bucket) of
+        BProps when is_list(BProps) ->
+            Type = proplists:get_value(datatype, BProps),
+            Mod = riak_kv_crdt:to_mod(Type),
+            supported(Mod);
+        {error, _}=Err ->
+            Err
+    end.
+
 %% @TODO in riak_dt change value to query allow query to take an
 %% argument, (so as to query subfields of map, or set membership etc)
 -spec crdt_value(module(), error | {ok, {dict(), crdt()}}) ->
@@ -492,6 +509,57 @@ get_context(Type, Value) ->
 %% ===================================================================
 -ifdef(TEST).
 
+is_crdt_test_() ->
+{setup,
+     fun() ->
+             meck:new(riak_core_bucket),
+             meck:new(riak_core_capability, []),
+             meck:expect(riak_core_capability, get,
+                         fun({riak_kv, crdt}, []) ->
+                                 [pncounter,riak_dt_pncounter,riak_dt_orswot,
+                                  riak_dt_map];
+                            (X, Y) -> meck:passthrough([X, Y]) end),
+             ok
+     end,
+     fun(_) ->
+             meck:unload(riak_core_capability),
+             meck:unload(riak_core_bucket)
+     end,
+     [
+      ?_test(begin
+                 meck:expect(riak_core_bucket, get_bucket,
+                             fun(_Bucket) -> [{datatype, foo}] end),
+                 Bucket = {<<"counterz">>, <<"crdt">>},
+                 BTProps = riak_core_bucket:get_bucket(Bucket),
+                 ?assertEqual(foo, proplists:get_value(datatype, BTProps)),
+                 ?assertNot(is_crdt(riak_object:new(Bucket, <<"k1">>, hello)))
+             end),
+      ?_test(begin
+                 Bucket = {<<"t">>, <<"bucketjumpy">>},
+                 ?assertNot(is_crdt(riak_object:new(Bucket, <<"k1">>, hi)))
+             end),
+      ?_test(begin
+                 meck:expect(riak_core_bucket, get_bucket,
+                             fun({<<"maps">>, _Name}) -> [{datatype, map}];
+                                ({<<"sets">>, _Name}) -> [{datatype, set}];
+                                ({<<"counters">>, _Name}) -> [{datatype, counter}];
+                                ({X, Y}) -> meck:passthrough([X, Y]) end),
+                 Bucket1 = {<<"maps">>, <<"crdt">>},
+                 Bucket2 = {<<"sets">>, <<"crdt">>},
+                 Bucket3 = {<<"counters">>, <<"crdt">>},
+                 BTPropsMap = riak_core_bucket:get_bucket(Bucket1),
+                 BTPropsSet = riak_core_bucket:get_bucket(Bucket2),
+                 BTPropsCounter = riak_core_bucket:get_bucket(Bucket3),
+                 ?assertEqual(map, proplists:get_value(datatype, BTPropsMap)),
+                 ?assertEqual(set, proplists:get_value(datatype, BTPropsSet)),
+                 ?assertEqual(counter,
+                              proplists:get_value(datatype, BTPropsCounter)),
+                 [?assert(is_crdt(riak_object:new(B, K, V)))
+                  || {B, K, V} <- [{Bucket1, <<"k1">>, hi},
+                                 {Bucket2, <<"k2">>, hey},
+                                 {Bucket3, <<"k3">>, hey}]]
+
+             end)]}.
 
 -ifdef(EQC).
 -define(QC_OUT(P),
@@ -507,13 +575,13 @@ eqc_test_() ->
      ?_test(?TIMED_QC(prop_binary_roundtrip()))}.
 
 prop_binary_roundtrip() ->
-    ?FORALL({_Type, Mod}, oneof(?MOD_MAP), 
+    ?FORALL({_Type, Mod}, oneof(?MOD_MAP),
             begin
                 {ok, ?CRDT{mod=SMod, value=SValue}} = from_binary(to_binary(?CRDT{mod=Mod, value=Mod:new()})),
                 conjunction([{module, equals(Mod, SMod)},
                              {value, Mod:equal(SValue, Mod:new())}])
             end).
-                
+
 
 -endif.
 -endif.