Merge pull request #706 from basho/bugfix/im/aae_fullsync_bucket_types_2.0

Backport Bugfix/im/aae_fullsync_bucket_types to 2.0

Reviewed-by: jonmeredith

Author: borshop

src/riak_repl2_fssink.erl

@@ -148,8 +148,9 @@ 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
-        {fs_diff_obj, RObj} ->
+    case binary_to_term(Data) of
+        {fs_diff_obj, BinObj} ->
+            RObj = riak_repl_util:decode_bin_obj(BinObj),
             riak_repl_util:do_repl_put(RObj);
         Other ->
             gen_fsm:send_event(State#state.fullsync_worker, Other)
@@ -185,18 +186,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_repl_fullsync_worker.erl

@@ -140,7 +140,7 @@ handle_cast({put, RObj, Pool}, State) ->
     poolboy:checkin(Pool, self()),
     {noreply, State};
 handle_cast({bin_put, BinObj, Pool}, State) ->
-    RObj = riak_repl_util:from_wire(BinObj),
+    RObj = riak_repl_util:decode_bin_obj(BinObj),
     riak_repl_util:do_repl_put(RObj),
     poolboy:checkin(Pool, self()), % resume work
     {noreply, State};

src/riak_repl_util.erl

@@ -56,6 +56,7 @@
          sockname/2,
          deduce_wire_version_from_proto/1,
          encode_obj_msg/2,
+         decode_bin_obj/1,
          make_pg_proxy_name/1,
          make_pg_name/1,
          mode_12_enabled/1,
@@ -875,28 +876,24 @@ 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}});
+    term_to_binary({Cmd, {BTHash, encode_obj(V, RObj)}}).
 
-        _W ->
-            BObj = riak_repl_util:to_wire(w1,RObj),
-            term_to_binary({Cmd, {BTHash, BObj}})
-    end.
+%% 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_bin_obj({BTHash, BinObj}) -> {BTHash, from_wire(BinObj)};
+decode_bin_obj(BinObj) -> from_wire(BinObj).
 
 %% @doc Create binary wire formatted replication blob for riak 2.0+, complete with
 %%      possible type, bucket and key for reconstruction on the other end. BinObj should be