emqx_batch_SUITE.erl 2.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657
  1. %%--------------------------------------------------------------------
  2. %% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
  3. %%
  4. %% Licensed under the Apache License, Version 2.0 (the "License");
  5. %% you may not use this file except in compliance with the License.
  6. %% You may obtain a copy of the License at
  7. %%
  8. %% http://www.apache.org/licenses/LICENSE-2.0
  9. %%
  10. %% Unless required by applicable law or agreed to in writing, software
  11. %% distributed under the License is distributed on an "AS IS" BASIS,
  12. %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. %% See the License for the specific language governing permissions and
  14. %% limitations under the License.
  15. %%--------------------------------------------------------------------
  16. -module(emqx_batch_SUITE).
  17. -compile(export_all).
  18. -compile(nowarn_export_all).
  19. -include_lib("eunit/include/eunit.hrl").
  20. all() -> emqx_ct:all(?MODULE).
  21. t_batch_full_commit(_) ->
  22. B0 = emqx_batch:init(#{batch_size => 3,
  23. linger_ms => 2000,
  24. commit_fun => fun(_) -> ok end
  25. }),
  26. B3 = lists:foldl(fun(E, B) -> emqx_batch:push(E, B) end, B0, [a, b, c]),
  27. ?assertEqual(3, emqx_batch:size(B3)),
  28. ?assertEqual([a, b, c], emqx_batch:items(B3)),
  29. %% Trigger commit fun.
  30. B4 = emqx_batch:push(a, B3),
  31. ?assertEqual(0, emqx_batch:size(B4)),
  32. ?assertEqual([], emqx_batch:items(B4)).
  33. t_batch_linger_commit(_) ->
  34. CommitFun = fun(Q) -> ?assertEqual(3, length(Q)) end,
  35. B0 = emqx_batch:init(#{batch_size => 3,
  36. linger_ms => 500,
  37. commit_fun => CommitFun
  38. }),
  39. B3 = lists:foldl(fun(E, B) -> emqx_batch:push(E, B) end, B0, [a, b, c]),
  40. ?assertEqual(3, emqx_batch:size(B3)),
  41. ?assertEqual([a, b, c], emqx_batch:items(B3)),
  42. receive
  43. batch_linger_expired ->
  44. B4 = emqx_batch:commit(B3),
  45. ?assertEqual(0, emqx_batch:size(B4)),
  46. ?assertEqual([], emqx_batch:items(B4))
  47. after
  48. 1000 ->
  49. error(linger_timer_not_triggered)
  50. end.