Merge pull request #665 from basho/bugfix/im/aae_fullsync_bucket_types

Update AAE fullsync to support bucket types

Author: bsparrow435

src/riak_repl2_fssink.erl

@@ -148,7 +148,7 @@ handle_info({Proto, Socket, Data},
         State=#state{socket=Socket,transport=Transport}) when Proto==tcp; Proto==ssl ->
     %% aae strategy will not receive messages here
     Transport:setopts(Socket, [{active, once}]),
-    case decode_obj_msg(Data) of
+    case riak_repl_util:decode_obj_msg(Data) of
         {fs_diff_obj, RObj} ->
             riak_repl_util:do_repl_put(RObj);
         Other ->
@@ -185,18 +185,6 @@ handle_info(init_ack, State=#state{socket=Socket,
 handle_info(_Msg, State) ->
     {noreply, State}.
 
-decode_obj_msg(Data) ->
-    Msg = binary_to_term(Data),
-    case Msg of
-        {fs_diff_obj, BObj} when is_binary(BObj) ->
-            RObj = riak_repl_util:from_wire(BObj),
-            {fs_diff_obj, RObj};
-        {fs_diff_obj, _RObj} ->
-            Msg;
-        Other ->
-            Other
-    end.
-
 terminate(_Reason, #state{fullsync_worker=FSW, work_dir=WorkDir, strategy=Strategy}) ->
     %% TODO: define a fullsync worker behavior and call it's stop function
     case is_pid(FSW) andalso is_process_alive(FSW) of

src/riak_repl2_fssink_pool.erl

@@ -2,7 +2,7 @@
 %% fullsyncs are running.
 
 -module(riak_repl2_fssink_pool).
--export([start_link/0, status/0, bin_put/1]).
+-export([start_link/0, status/0, put/1, bin_put/1]).
 
 start_link() ->
     MinPool = app_helper:get_env(riak_repl, fssink_min_workers, 5),
@@ -21,6 +21,12 @@ status() ->
      {overflow, Overflow},
      {num_monitors, NumMonitors}].
 
+%% @doc Send an object or BT hash / object pair to the worker pool to
+%% run a put against. No guarantees of completion.
+put(Obj) ->
+    Pid = poolboy:checkout(?MODULE, true, infinity),
+    riak_repl_fullsync_worker:do_put(Pid, Obj, ?MODULE).
+
 %% @doc Send a replication wire-encoded binary to the worker pool
 %% for running a put against.  No guarantees of completion.
 bin_put(BinObj) ->

src/riak_repl_aae_sink.erl

@@ -83,7 +83,7 @@ handle_info({Proto, _Socket, Data}, State=#state{transport=Transport,
         [MsgType|<<>>] ->
             process_msg(MsgType, State);
         [MsgType|MsgData] ->
-            process_msg(MsgType, binary_to_term(MsgData), State)
+            process_msg(MsgType, riak_repl_util:decode_obj_msg(MsgData), State)
     end;
 
 handle_info({'DOWN', _, _, _, _}, State) ->
@@ -143,10 +143,10 @@ process_msg(?MSG_GET_AAE_SEGMENT, {SegmentNum,IndexN}, State=#state{tree_pid=Tre
     send_reply(ResponseMsg, State);
 
 %% no reply
-process_msg(?MSG_PUT_OBJ, {fs_diff_obj, BObj}, State) ->
+process_msg(?MSG_PUT_OBJ, {fs_diff_obj, Obj}, State) ->
     %% may block on worker pool, ok return means work was submitted
     %% to pool, not that put FSM completed successfully.
-    ok = riak_repl2_fssink_pool:bin_put(BObj),
+    ok = riak_repl2_fssink_pool:put(Obj),
     {noreply, State};
 
 %% replies: ok | not_responsible

src/riak_repl_util.erl

@@ -55,6 +55,7 @@
          sockname/2,
          deduce_wire_version_from_proto/1,
          encode_obj_msg/2,
+         decode_obj_msg/1,
          make_pg_proxy_name/1,
          make_pg_name/1,
          mode_12_enabled/1,
@@ -867,27 +868,35 @@ encode_obj_msg(V, {Cmd, RObj}) ->
     encode_obj_msg(V, {Cmd, RObj}, riak_object:type(RObj)).
 
 encode_obj_msg(V, {Cmd, RObj}, undefined) ->
-    case V of
-        w0 ->
-            term_to_binary({Cmd, RObj});
-        _W ->
-            BObj = riak_repl_util:to_wire(w1,RObj),
-            term_to_binary({Cmd, BObj})
-    end;
+    term_to_binary({Cmd, encode_obj(V, RObj)});
 encode_obj_msg(V, {Cmd, RObj}, T) ->
     BTHash = case riak_repl_bucket_type_util:property_hash(T) of
                  undefined ->
                      0;
                  Hash ->
                      Hash
              end,
-    case V of
-        w0 ->
-            term_to_binary({Cmd, {BTHash, RObj}});
-
-        _W ->
-            BObj = riak_repl_util:to_wire(w1,RObj),
-            term_to_binary({Cmd, {BTHash, BObj}})
+    term_to_binary({Cmd, {BTHash, encode_obj(V, RObj)}}).
+
+%% A wrapper around to_wire which leaves the object unencoded when using the w0 wire protocol.
+encode_obj(w0, RObj) ->
+    RObj;
+encode_obj(W, RObj) ->
+    to_wire(W, RObj).
+
+decode_obj_msg(Data) ->
+    Msg = binary_to_term(Data),
+    case Msg of
+        {fs_diff_obj, BObj} when is_binary(BObj) ->
+            RObj = from_wire(BObj),
+            {fs_diff_obj, RObj};
+        {fs_diff_obj, {BTHash, BObj}} when is_binary(BObj) ->
+            RObj = from_wire(BObj),
+            {fs_diff_obj, {BTHash, RObj}};
+        {fs_diff_obj, _} ->
+            Msg;
+        Other ->
+            Other
     end.
 
 %% @doc Create binary wire formatted replication blob for riak 2.0+, complete with