DC-97: implement PullRange interface (#82)

* DC-97: implement PullRange interface

* bump dmt_client and fix tests

rebar.lock

@@ -8,11 +8,11 @@
  {<<"cowlib">>,{pkg,<<"cowlib">>,<<"1.0.2">>},2},
  {<<"dmsl">>,
   {git,"git@github.com:rbkmoney/damsel.git",
-       {ref,"21bd2877a0c9d0ee7c2ed0930086f46f7a08223d"}},
+       {ref,"50a0d4f6df546fce4d16617b796f031703dd4707"}},
   0},
  {<<"dmt_client">>,
   {git,"git@github.com:rbkmoney/dmt_client.git",
-       {ref,"8387fbed6f9ae4f00e5442c611589d321db55b90"}},
+       {ref,"dca3a0a9ee864a845f9b4571e69038d3acb25bd2"}},
   0},
  {<<"dmt_core">>,
   {git,"git@github.com:rbkmoney/dmt_core.git",

src/dmt_api_repository.erl

@@ -6,19 +6,21 @@
 
 -export([checkout/3]).
 -export([checkout_object/4]).
--export([pull/3]).
+-export([pull/4]).
 -export([commit/4]).
 
 -export_type([version/0]).
+-export_type([limit/0]).
 -export_type([snapshot/0]).
 -export_type([commit/0]).
 -export_type([history/0]).
 -export_type([operation_conflict/0]).
 
--type version() :: dmsl_domain_config_thrift:'Version'().
+-type version()  :: dmsl_domain_config_thrift:'Version'().
+-type limit()    :: dmsl_domain_config_thrift:'Limit'() | undefined.
 -type snapshot() :: dmsl_domain_config_thrift:'Snapshot'().
--type commit() :: dmsl_domain_config_thrift:'Commit'().
--type history() :: dmsl_domain_config_thrift:'History'().
+-type commit()   :: dmsl_domain_config_thrift:'Commit'().
+-type history()  :: dmsl_domain_config_thrift:'History'().
 
 -type ref() :: dmsl_domain_config_thrift:'Reference'().
 -type object_ref() :: dmsl_domain_thrift:'Reference'().
@@ -35,7 +37,7 @@
     % TODO this was made due to dialyzer warning, can't find the way to fix it
     {ok, term()} |
     {error, version_not_found}.
--callback pull(version(), context()) ->
+-callback pull(version(), limit(),  context()) ->
     {ok, history()} |
     {error, version_not_found}.
 -callback commit(version(), commit(), context()) ->
@@ -79,11 +81,11 @@ checkout_object(Reference, ObjectReference, Repository, Context) ->
             Error
     end.
 
--spec pull(version(), repository(), context()) ->
+-spec pull(version(), limit(), repository(), context()) ->
     {ok, history()} | {error, version_not_found}.
 
-pull(Version, Repository, Context) ->
-    Repository:pull(Version, Context).
+pull(Version, Limit, Repository, Context) ->
+    Repository:pull(Version, Limit, Context).
 
 -spec commit(version(), commit(), repository(), context()) ->
     {ok, version()} |

src/dmt_api_repository_handler.erl

@@ -14,6 +14,8 @@
         {ok, dmt_api_repository:version()} | no_return();
     ('Checkout', woody:args(), context(), woody:options()) ->
         {ok, dmt_api_repository:snapshot()} | no_return();
+    ('PullRange', woody:args(), context(), woody:options()) ->
+        {ok, dmt_api_repository:history()} | no_return();
     ('Pull', woody:args(), context(), woody:options()) ->
         {ok, dmt_api_repository:history()} | no_return().
 handle_function('Commit', [Version, Commit], Context, Repository) ->
@@ -38,8 +40,16 @@ handle_function('Checkout', [Reference], Context, Repository) ->
         {error, version_not_found} ->
             woody_error:raise(business, #'VersionNotFound'{})
     end;
+handle_function('PullRange', [After, Limit], Context, Repository) ->
+    case dmt_api_repository:pull(After, Limit, Repository, Context) of
+        {ok, History} ->
+            {ok, History};
+        {error, version_not_found} ->
+            woody_error:raise(business, #'VersionNotFound'{})
+    end;
+%% depreceted, will be removed soon
 handle_function('Pull', [Version], Context, Repository) ->
-    case dmt_api_repository:pull(Version, Repository, Context) of
+    case dmt_api_repository:pull(Version, undefined, Repository, Context) of
         {ok, History} ->
             {ok, History};
         {error, version_not_found} ->

src/dmt_api_repository_migration.erl

@@ -16,6 +16,7 @@
 
 -export([checkout/2]).
 -export([pull/2]).
+-export([pull/3]).
 -export([commit/3]).
 
 %% State processor
@@ -51,11 +52,18 @@ checkout(Ref, Context) ->
     {error, version_not_found}.
 
 pull(Version, Context) ->
+        pull(Version, undefined, Context).
+
+-spec pull(dmt_api_repository:version(), dmt_api_repository:limit(), context()) ->
+    {ok, dmt_api_repository:history()} |
+    {error, version_not_found}.
+
+pull(Version, Limit, Context) ->
     case is_migration_finished(Context) of
         true ->
-            dmt_api_repository_v4:pull(Version, Context);
+            dmt_api_repository_v4:pull(Version, Limit, Context);
         false ->
-            dmt_api_repository_v3:pull(Version, Context)
+            dmt_api_repository_v3:pull(Version, Limit, Context)
     end.
 
 -spec commit(dmt_api_repository:version(), commit(), context()) ->

src/dmt_api_repository_v4.erl

@@ -13,6 +13,7 @@
 
 -export([checkout/2]).
 -export([pull/2]).
+-export([pull/3]).
 -export([commit/3]).
 
 %% State processor
@@ -78,8 +79,15 @@ checkout({version, V}, Context) ->
     {error, version_not_found}.
 
 pull(Version, Context) ->
+    pull(Version, undefined, Context).
+
+-spec pull(dmt_api_repository:version(), dmt_api_repository:limit(), context()) ->
+    {ok, dmt_api_repository:history()} |
+    {error, version_not_found}.
+
+pull(Version, Limit, Context) ->
     After = get_event_id(Version),
-    case get_history_by_range(#mg_stateproc_HistoryRange{'after' = After}, Context) of
+    case get_history_by_range(#mg_stateproc_HistoryRange{'after' = After, 'limit' = Limit}, Context) of
         #st{history = History} ->
             {ok, History};
         {error, version_not_found} ->

test/dmt_api_tests_SUITE.erl

@@ -24,6 +24,7 @@
 -type config() :: [{atom(), term()}].
 
 -define(config(Key, C), (element(2, lists:keyfind(Key, 1, C)))).
+-define(DEFAULT_LIMIT, 9001). % to emulate unlimited polling
 
 -type test_case_name() :: atom().
 -type group_name() :: atom().
@@ -79,6 +80,7 @@ init_per_suite(C) ->
         ]) ++
         genlib_app:start_application_with(dmt_client, [
             {cache_update_interval, 5000}, % milliseconds
+            {cache_update_pull_limit, ?DEFAULT_LIMIT},
             {max_cache_size, #{
                 elements => 20,
                 memory => 52428800 % 50Mb
@@ -186,12 +188,12 @@ delete(_C) ->
 -spec pull_commit(term()) -> term().
 pull_commit(_C) ->
     ID = next_id(),
-    History1 = #{} = dmt_client:pull(0),
+    History1 = #{} = dmt_client:pull_range(0, ?DEFAULT_LIMIT),
     Version1 = lists:max([0 | maps:keys(History1)]),
     Object = fixture_domain_object(ID, <<"PullFixture">>),
     Commit = #'Commit'{ops = [{insert, #'InsertOp'{object = Object}}]},
     Version2 = dmt_client:commit(Version1, Commit),
-    #{Version2 := Commit} = dmt_client:pull(Version1).
+    #{Version2 := Commit} = dmt_client:pull_range(Version1, ?DEFAULT_LIMIT).
 
 -spec retry_commit(term()) -> term().
 retry_commit(_C) ->