Merge pull request #401 from basho/mas-i1788-reconcileapi

Mas i1788 reconcileapi

Author: martinsumner

rebar.config

@@ -17,7 +17,7 @@
 ]}.
 
 {deps, [
-    {riak_pb, {git, "https://github.com/basho/riak_pb", {tag, "3.0.4"}}}
+    {riak_pb, {git, "https://github.com/basho/riak_pb", {branch, "develop-3.0"}}}
 ]}.
 
 {edoc_opts, [

src/riakc_pb_socket.erl

@@ -49,7 +49,7 @@
          set_client_id/2, set_client_id/3,
          get_server_info/1, get_server_info/2,
          get/3, get/4, get/5,
-         fetch/2,
+         fetch/2, fetch/3, push/3,
          put/2, put/3, put/4,
          delete/3, delete/4, delete/5,
          delete_vclock/4, delete_vclock/5, delete_vclock/6,
@@ -338,14 +338,45 @@ get(Pid, Bucket, Key, Options, Timeout) ->
     Req = get_options(Options, #rpbgetreq{type =T, bucket = B, key = Key}),
     call_infinity(Pid, {req, Req, Timeout}).
 
+
 %% @doc Fetch replicated objects from a queue
 -spec fetch(pid(), binary()) ->
                 {ok, queue_empty}|
-                {ok|crc_wonky, {deleted, term(), binary()}|binary()}.
+                {ok|crc_wonky,
+                    {deleted, term(), binary()}|binary()}.
 fetch(Pid, QueueName) ->
-    Req = #rpbfetchreq{queuename = QueueName},
+    fetch(Pid, QueueName, internal).
+
+%% @doc Fetch with specific format may also return segment ID and hash
+-spec fetch(pid(), binary(), internal|internal_aaehash) ->
+                {ok, queue_empty}|
+                {ok|crc_wonky,
+                    {deleted, term(), binary()}|
+                        binary()|
+                        {deleted, term(), binary(), non_neg_integer(), non_neg_integer()}|
+                        {binary(), non_neg_integer(), non_neg_integer()}}.
+fetch(Pid, QueueName, ObjectFormat)
+        when ObjectFormat =:= internal; ObjectFormat =:= internal_aaehash ->
+    Req = #rpbfetchreq{queuename = QueueName,
+                        encoding = erlang:atom_to_binary(ObjectFormat, utf8)},
     call_infinity(Pid, {req, Req, default_timeout(get_timeout)}).
 
+%% @doc Push to nextgenrepl replication queue a list of Buckets/Keys/Clocks
+%% where the push will occur if the object is fetched an is currently at that
+%% clock
+-spec push(pid(),
+            binary(),
+            [{riakc_obj:bucket(), riakc_obj:key(), riakc_obj:vclock()}]) ->
+                {error, term()}|{ok, iolist()}.
+push(Pid, QueueName, BucketKeyClockList) ->
+    KeysValue = lists:map(fun make_keyvalue/1, BucketKeyClockList),
+    Req = #rpbpushreq{queuename = QueueName, keys_value = KeysValue},
+    call_infinity(Pid, {req, Req, default_timeout(get_timeout)}).
+
+make_keyvalue({{T, B}, K, C}) ->
+    #rpbkeysvalue{type = T, bucket = B, key = K, value = C};
+make_keyvalue({B, K, C}) ->
+    #rpbkeysvalue{bucket = B, key = K, value = C}.
 
 %% @doc Put the metadata/value in the object under bucket/key
 %% @equiv put(Pid, Obj, [])
@@ -2382,14 +2413,55 @@ process_response(#request{msg = #rpbfetchreq{}},
                  #rpbfetchresp{deleted = true, 
                                 crc_check = CRC,
                                 replencoded_object = ObjBin,
-                                deleted_vclock = VclockBin}, State) ->
+                                deleted_vclock = VclockBin,
+                                segment_id = undefined,
+                                segment_hash = undefined}, State) ->
     {reply,
         {crc_check(CRC,ObjBin), {deleted, VclockBin, ObjBin}},
         State};
+process_response(#request{msg = #rpbfetchreq{}},
+                 #rpbfetchresp{deleted = true, 
+                                crc_check = CRC,
+                                replencoded_object = ObjBin,
+                                deleted_vclock = VclockBin,
+                                segment_id = SegID,
+                                segment_hash = SegHash}, State) ->
+    {reply,
+        {crc_check(CRC,ObjBin), {deleted, VclockBin, ObjBin, SegID, SegHash}},
+        State};
 process_response(#request{msg = #rpbfetchreq{}},
                  #rpbfetchresp{crc_check = CRC,
-                                replencoded_object = ObjBin}, State) ->
+                                replencoded_object = ObjBin,
+                                segment_id = undefined,
+                                segment_hash = undefined}, State) ->
     {reply, {crc_check(CRC,ObjBin), ObjBin}, State};
+process_response(#request{msg = #rpbfetchreq{}},
+                 #rpbfetchresp{crc_check = CRC,
+                                replencoded_object = ObjBin,
+                                segment_id = SegID,
+                                segment_hash = SegHash}, State) ->
+    {reply, {crc_check(CRC,ObjBin), {ObjBin, SegID, SegHash}}, State};
+
+%% rpbpushreq
+process_response(#request{msg = #rpbpushreq{queuename = Q}}, 
+                    #rpbpushresp{queue_exists = true,
+                                    queuename = Q,
+                                    foldq_length = FL,
+                                    fsync_length = FSL,
+                                    realt_length = RTL}, State) ->
+    {reply,
+        {ok, 
+            iolist_to_binary(
+                io_lib:format("Queue ~s: ~w ~w ~w", [Q, FL, FSL, RTL]))},
+        State};
+process_response(#request{msg = #rpbpushreq{queuename = Q}}, 
+                    #rpbpushresp{queue_exists = false}, State) ->
+    {reply,
+        {ok, 
+            iolist_to_binary(io_lib:format("No queue ~s", [Q]))},
+        State};
+
+
 
 %% rpbputreq
 process_response(#request{msg = #rpbputreq{}},