Merge pull request #81 from basho/mas-i1775-fetchclocksmodrange

Mas i1775 fetchclocksmodrange

Author: martinsumner

include/rhc.hrl

@@ -21,6 +21,7 @@
 %% -------------------------------------------------------------------
 
 -define(DEFAULT_TIMEOUT, 60000).
+-define(AAEFOLD_TIMEOUT, 3600000).
 
 -record(rhc, {ip,
               port,

rebar.config

@@ -12,7 +12,7 @@
    {mochiweb, {git, "git://github.com/basho/mochiweb.git", {tag, "riak_kv-3.0.0"}}},
 
    %% riak-erlang-client for riakc_obj
-   {riakc, {git, "git://github.com/basho/riak-erlang-client", {tag, "riak_kv-3.0.1"}}}
+   {riakc, {git, "git://github.com/basho/riak-erlang-client", {branch, "develop-3.0"}}}
   ]}.
 {edoc_opts,
  [

src/rhc.erl

@@ -69,6 +69,7 @@
          aae_merge_root/2,
          aae_merge_branches/3,
          aae_fetch_clocks/3,
+         aae_fetch_clocks/4,
          aae_range_tree/7,
          aae_range_clocks/5,
          aae_range_replkeys/5,
@@ -299,7 +300,7 @@ rt_enqueue(Rhc, Bucket, Key, Options) ->
 aae_merge_root(Rhc, NVal) ->
     Url = make_cached_aae_url(Rhc, root, NVal, undefined),
 
-    case request(get, Url, ["200"], [], [], Rhc) of
+    case request(get, Url, ["200"], [], [], Rhc, ?AAEFOLD_TIMEOUT) of
         {ok, _Status, _Headers, Body} ->
             {struct, Response} = mochijson2:decode(Body),
             {ok, erlify_aae_root(Response)};
@@ -317,16 +318,16 @@ aae_merge_root(Rhc, NVal) ->
 aae_merge_branches(Rhc, NVal, Branches) ->
     Url = make_cached_aae_url(Rhc, branch, NVal, Branches),
 
-    case request(get, Url, ["200"], [], [], Rhc) of
+    case request(get, Url, ["200"], [], [], Rhc, ?AAEFOLD_TIMEOUT) of
         {ok, _Status, _Headers, Body} ->
             {struct, Response} = mochijson2:decode(Body),
             {ok, erlify_aae_branches(Response)};
         {error, Error} ->
             {error, Error}
     end.
 
-%% @doc get the aae merged branches for the given `NVal', restricted
-%% to the given list of `Branches'
+%% @doc fetch the keys and clocks for the given `NVal', restricted
+%% to the given list of `Segments'
 -spec aae_fetch_clocks(rhc(),
                        NVal::pos_integer(),
                        Segments::list(pos_integer())) ->
@@ -335,7 +336,26 @@ aae_merge_branches(Rhc, NVal, Branches) ->
 aae_fetch_clocks(Rhc, NVal, Segments) ->
     Url = make_cached_aae_url(Rhc, keysclocks, NVal, Segments),
 
-    case request(get, Url, ["200"], [], [], Rhc) of
+    case request(get, Url, ["200"], [], [], Rhc, ?AAEFOLD_TIMEOUT) of
+        {ok, _Status, _Headers, Body} ->
+            {struct, Response} = mochijson2:decode(Body),
+            {ok, erlify_aae_keysclocks(Response)};
+        {error, Error} ->
+            {error, Error}
+    end.
+
+%% @doc fetch the keys and clocks for the given `NVal', restricted
+%% to the given list of `Segments' and by a modified range
+-spec aae_fetch_clocks(rhc(),
+                       NVal::pos_integer(),
+                       Segments::list(pos_integer()),
+                       ModifiedRange::modified_range()) ->
+                              {ok, {keysclocks, [{{riakc_obj:bucket(), riakc_obj:key()}, binary()}]}} |
+                              {error, any()}.
+aae_fetch_clocks(Rhc, NVal, Segments, ModifiedRange) ->
+    Url =
+        make_cached_aae_url(Rhc, keysclocks, NVal, {Segments, ModifiedRange}),
+    case request(get, Url, ["200"], [], [], Rhc, ?AAEFOLD_TIMEOUT) of
         {ok, _Status, _Headers, Body} ->
             {struct, Response} = mochijson2:decode(Body),
             {ok, erlify_aae_keysclocks(Response)};
@@ -381,7 +401,7 @@ aae_range_tree(Rhc, BucketAndType, KeyRange,
            "?filter=", encode_aae_range_filter(KeyRange, SegmentFilter, ModifiedRange, HashMethod)
           ]),
 
-    case request(get, Url, ["200"], [], [], Rhc) of
+    case request(get, Url, ["200"], [], [], Rhc, ?AAEFOLD_TIMEOUT) of
         {ok, _Status, _Headers, Body} ->
             {struct, Response} = mochijson2:decode(Body),
             {ok, erlify_aae_tree(Response)};
@@ -404,7 +424,7 @@ aae_range_clocks(Rhc, BucketAndType, KeyRange, SegmentFilter, ModifiedRange) ->
            "?filter=", encode_aae_range_filter(KeyRange, SegmentFilter, ModifiedRange, undefined)
           ]),
 
-    case request(get, Url, ["200"], [], [], Rhc) of
+    case request(get, Url, ["200"], [], [], Rhc, ?AAEFOLD_TIMEOUT) of
         {ok, _Status, _Headers, Body} ->
             {struct, Response} = mochijson2:decode(Body),
             {ok, erlify_aae_keysclocks(Response)};
@@ -436,7 +456,7 @@ aae_range_replkeys(Rhc, BucketType, KeyRange, ModifiedRange, QueueName) ->
             encode_aae_range_filter(KeyRange, all, ModifiedRange, undefined)
           ]),
 
-    case request(get, Url, ["200"], [], [], Rhc) of
+    case request(get, Url, ["200"], [], [], Rhc, ?AAEFOLD_TIMEOUT) of
         {ok, _Status, _Headers, Body} ->
             {struct, Response} = mochijson2:decode(Body),
             [{<<"dispatched_count">>, DispatchedCount}] = Response,
@@ -486,7 +506,7 @@ aae_find_keys(Rhc, BucketAndType, KeyRange, ModifiedRange, Query) ->
            "?filter=", encode_aae_find_keys_filter(KeyRange, undefined, ModifiedRange)
           ]),
 
-    case request(get, Url, ["200"], [], [], Rhc) of
+    case request(get, Url, ["200"], [], [], Rhc, ?AAEFOLD_TIMEOUT) of
         {ok, _Status, _Headers, Body} ->
             {struct, Response} = mochijson2:decode(Body),
             {ok, erlify_aae_find_keys(Response)};
@@ -520,7 +540,7 @@ aae_find_tombs(Rhc, BucketAndType, KeyRange, SegmentFilter, ModifiedRange) ->
                                                 ModifiedRange)
           ]),
 
-    case request(get, Url, ["200"], [], [], Rhc) of
+    case request(get, Url, ["200"], [], [], Rhc, ?AAEFOLD_TIMEOUT) of
         {ok, _Status, _Headers, Body} ->
             {struct, Response} = mochijson2:decode(Body),
             {ok, erlify_aae_find_keys(Response)};
@@ -566,7 +586,7 @@ aae_reap_tombs(Rhc,
                                             ChangeMethod)
           ]),
 
-    case request(get, Url, ["200"], [], [], Rhc) of
+    case request(get, Url, ["200"], [], [], Rhc, ?AAEFOLD_TIMEOUT) of
         {ok, _Status, _Headers, Body} ->
             {struct, Response} = mochijson2:decode(Body),
             [{<<"dispatched_count">>, DispatchedCount}] = Response,
@@ -613,7 +633,7 @@ aae_erase_keys(Rhc,
                                             ChangeMethod)
           ]),
 
-    case request(get, Url, ["200"], [], [], Rhc) of
+    case request(get, Url, ["200"], [], [], Rhc, ?AAEFOLD_TIMEOUT) of
         {ok, _Status, _Headers, Body} ->
             {struct, Response} = mochijson2:decode(Body),
             [{<<"dispatched_count">>, DispatchedCount}] = Response,
@@ -655,7 +675,7 @@ aae_object_stats(Rhc, BucketAndType, KeyRange, ModifiedRange) ->
            "?filter=", encode_aae_find_keys_filter(KeyRange, undefined, ModifiedRange)
           ]),
 
-    case request(get, Url, ["200"], [], [], Rhc) of
+    case request(get, Url, ["200"], [], [], Rhc, ?AAEFOLD_TIMEOUT) of
         {ok, _Status, _Headers, Body} ->
             {struct, Response} = mochijson2:decode(Body),
             {ok, {stats, erlify_aae_object_stats(Response)}};
@@ -681,7 +701,7 @@ aae_list_buckets(Rhc, MinNVal) when is_integer(MinNVal), MinNVal > 0 ->
                             "?filter=", integer_to_list(MinNVal)]),
     aae_list_buckets(Rhc, Url);
 aae_list_buckets(Rhc, Url) when is_list(Url) ->
-    case request(get, Url, ["200"], [], [], Rhc) of
+    case request(get, Url, ["200"], [], [], Rhc, ?AAEFOLD_TIMEOUT) of
         {ok, _Status, _Headers, Body} ->
             {struct, [{<<"results">>, Response}]} = mochijson2:decode(Body),
             {ok, erlify_aae_buckets(Response)};
@@ -1355,25 +1375,51 @@ make_rtenqueue_url(Rhc=#rhc{}, BucketAndType, Key, Query) ->
 -spec make_cached_aae_url(rhc(),
                           root | branch | keysclocks,
                           NVal :: pos_integer(),
-                          Filter :: proplists:proplist()|undefined) ->
+                          IDs :: list(non_neg_integer())|
+                                    undefined|
+                                    {list(non_neg_integer()),
+                                        modified_range()}) ->
                                  iolist().
-make_cached_aae_url(Rhc, Type, NVal, Filter) ->
+make_cached_aae_url(Rhc, Type, NVal, undefined) ->
+    complete_cached_aae_url(Rhc, NVal, Type, []);
+make_cached_aae_url(Rhc, Type, NVal, IDs) when is_list(IDs) ->
+    Filter = ["?filter=", encode_aae_cached_filter(IDs)],
+    complete_cached_aae_url(Rhc, NVal, Type, Filter);
+make_cached_aae_url(Rhc, Type, NVal, {IDs, ModifiedRange}) ->
+    Filter = ["?filter=", encode_aae_cached_filter(IDs, ModifiedRange)],
+    complete_cached_aae_url(Rhc, NVal, Type, Filter).
+
+
+complete_cached_aae_url(Rhc, NVal, Type, Filter) ->
     lists:flatten(
       [root_url(Rhc),
        "cachedtrees", "/", %% the AAE-Fold cachedtrees prefix
        "nvals", "/",
        integer_to_list(NVal), "/",
        atom_to_list(Type),
-       [ ["?filter=", encode_aae_cached_filter(Filter)] || Filter =/= undefined]
+       Filter
       ]).
 
+
+
 %% @doc this is a list of integers. Segment IDs or Branches, but
 %% either way, just json encode a list of ints
 -spec encode_aae_cached_filter(list(pos_integer())) -> string().
 encode_aae_cached_filter(Filter) ->
     JSON = mochijson2:encode(Filter),
     base64:encode_to_string(lists:flatten(JSON)).
 
+-spec encode_aae_cached_filter(list(pos_integer()),
+                                modified_range()) -> string().
+encode_aae_cached_filter(SegmentIDs, ModifiedRange) ->
+    FilterElems = [EncodeFun(FilterElem) || {EncodeFun, FilterElem} <-
+                                  [{fun encode_segment_filter/1,
+                                      {SegmentIDs, large}},
+                                   {fun encode_modified_range/1,
+                                       ModifiedRange}]],
+    JSON = mochijson2:encode({struct, lists:flatten(FilterElems)}),
+    base64:encode_to_string(iolist_to_binary(JSON)).
+
 -spec encode_aae_find_keys_filter(key_range(),
                                     segment_filter() | undefined,
                                     modified_range()) ->
@@ -1483,11 +1529,15 @@ make_datatype_url(Rhc, BucketAndType, Key, Query) ->
 
 %% @doc send an ibrowse request
 request(Method, Url, Expect, Headers, Body, Rhc) ->
+    request(Method, Url, Expect, Headers, Body, Rhc, ?DEFAULT_TIMEOUT).
+
+request(Method, Url, Expect, Headers, Body, Rhc, Timeout) ->
     AuthHeader = get_auth_header(Rhc#rhc.options),
     SSLOptions = get_ssl_options(Rhc#rhc.options),
     Accept = {"Accept", "multipart/mixed, */*;q=0.9"},
     case ibrowse:send_req(Url, [Accept|Headers] ++ AuthHeader, Method, Body,
-                          [{response_format, binary}] ++ SSLOptions) of
+                          [{response_format, binary}] ++ SSLOptions,
+                          Timeout) of
         Resp={ok, Status, _, _} ->
             case lists:member(Status, Expect) of
                 true -> Resp;