bucket_reloader.erl

-module(bucket_reloader).

-export([reload/3]).

reload(FromServer, ToServer, Bucket) ->
    {ok, CFrom} = riak:client_connect(FromServer),
    {ok, CTo} = riak:client_connect(ToServer),
    {ok, Keys} = CFrom:list_keys(Bucket),
    io:format("Transferring ~p keys~n", [length(Keys)]),
    transfer(CFrom, CTo, Bucket, Keys, 0).

transfer(_CFrom, _CTo, _Bucket, [], _) ->
    io:format("~n"),
    ok;
transfer(CFrom, CTo, Bucket, [H|T], Count) when is_binary(H) ->
    Owner = self(),
    proc_lib:spawn(fun() ->
                           case CFrom:get(Bucket, H) of
                               {ok, FromObj} ->
                                   CTo:put(FromObj, 1),
                                   io:format("."),
                                   Owner ! done;
                               Error ->
                                   error_logger:error_msg("Error fetching ~p/~p: ~p~n", [Bucket, H, Error]),
                                   Owner ! done
                           end end),
    NewCount = if
                   Count == 250 ->
                       let_workers_catch_up(Count),
                       0;
                   true ->
                       Count + 1
               end,
    transfer(CFrom, CTo, Bucket, T, NewCount).

let_workers_catch_up(0) ->
    ok;
let_workers_catch_up(Count) ->
    receive
        done ->
            ok
    end,
    let_workers_catch_up(Count - 1).