emqx_mqueue_SUITE.erl 5.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165
  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_mqueue_SUITE).
  17. -compile(export_all).
  18. -compile(nowarn_export_all).
  19. -include("emqx.hrl").
  20. -include("emqx_mqtt.hrl").
  21. -include_lib("eunit/include/eunit.hrl").
  22. -define(Q, emqx_mqueue).
  23. all() -> emqx_ct:all(?MODULE).
  24. t_init(_) ->
  25. error('TODO').
  26. t_is_empty(_) ->
  27. error('TODO').
  28. t_len(_) ->
  29. error('TODO').
  30. t_max_len(_) ->
  31. error('TODO').
  32. t_dropped(_) ->
  33. error('TODO').
  34. t_stats(_) ->
  35. error('TODO').
  36. t_in(_) ->
  37. Opts = #{max_len => 5, store_qos0 => true},
  38. Q = ?Q:init(Opts),
  39. ?assert(?Q:is_empty(Q)),
  40. {_, Q1} = ?Q:in(#message{}, Q),
  41. ?assertEqual(1, ?Q:len(Q1)),
  42. {_, Q2} = ?Q:in(#message{qos = 1}, Q1),
  43. ?assertEqual(2, ?Q:len(Q2)),
  44. {_, Q3} = ?Q:in(#message{qos = 2}, Q2),
  45. {_, Q4} = ?Q:in(#message{}, Q3),
  46. {_, Q5} = ?Q:in(#message{}, Q4),
  47. ?assertEqual(5, ?Q:len(Q5)).
  48. t_in_qos0(_) ->
  49. Opts = #{max_len => 5, store_qos0 => false},
  50. Q = ?Q:init(Opts),
  51. {_, Q1} = ?Q:in(#message{qos = 0}, Q),
  52. ?assert(?Q:is_empty(Q1)),
  53. {_, Q2} = ?Q:in(#message{qos = 0}, Q1),
  54. ?assert(?Q:is_empty(Q2)).
  55. t_out(_) ->
  56. Opts = #{max_len => 5, store_qos0 => true},
  57. Q = ?Q:init(Opts),
  58. {empty, Q} = ?Q:out(Q),
  59. {_, Q1} = ?Q:in(#message{}, Q),
  60. {Value, Q2} = ?Q:out(Q1),
  61. ?assertEqual(0, ?Q:len(Q2)),
  62. ?assertEqual({value, #message{}}, Value).
  63. t_simple_mqueue(_) ->
  64. Opts = #{max_len => 3, store_qos0 => false},
  65. Q = ?Q:init(Opts),
  66. ?assertEqual(3, ?Q:max_len(Q)),
  67. ?assert(?Q:is_empty(Q)),
  68. {_, Q1} = ?Q:in(#message{qos = 1, payload = <<"1">>}, Q),
  69. {_, Q2} = ?Q:in(#message{qos = 1, payload = <<"2">>}, Q1),
  70. {_, Q3} = ?Q:in(#message{qos = 1, payload = <<"3">>}, Q2),
  71. {_, Q4} = ?Q:in(#message{qos = 1, payload = <<"4">>}, Q3),
  72. ?assertEqual(3, ?Q:len(Q4)),
  73. {{value, Msg}, Q5} = ?Q:out(Q4),
  74. ?assertEqual(<<"2">>, Msg#message.payload),
  75. ?assertEqual([{len, 2}, {max_len, 3}, {dropped, 1}], ?Q:stats(Q5)).
  76. t_infinity_simple_mqueue(_) ->
  77. Opts = #{max_len => 0, store_qos0 => false},
  78. Q = ?Q:init(Opts),
  79. ?assert(?Q:is_empty(Q)),
  80. ?assertEqual(0, ?Q:max_len(Q)),
  81. Qx = lists:foldl(
  82. fun(I, AccQ) ->
  83. {_, NewQ} = ?Q:in(#message{qos = 1, payload = iolist_to_binary([I])}, AccQ),
  84. NewQ
  85. end, Q, lists:seq(1, 255)),
  86. ?assertEqual(255, ?Q:len(Qx)),
  87. ?assertEqual([{len, 255}, {max_len, 0}, {dropped, 0}], ?Q:stats(Qx)),
  88. {{value, V}, _Qy} = ?Q:out(Qx),
  89. ?assertEqual(<<1>>, V#message.payload).
  90. t_priority_mqueue(_) ->
  91. Opts = #{max_len => 3,
  92. priorities =>
  93. #{<<"t1">> => 1,
  94. <<"t2">> => 2,
  95. <<"t3">> => 3
  96. },
  97. store_qos0 => false},
  98. Q = ?Q:init(Opts),
  99. ?assertEqual(3, ?Q:max_len(Q)),
  100. ?assert(?Q:is_empty(Q)),
  101. {_, Q1} = ?Q:in(#message{qos = 1, topic = <<"t2">>}, Q),
  102. {_, Q2} = ?Q:in(#message{qos = 1, topic = <<"t1">>}, Q1),
  103. {_, Q3} = ?Q:in(#message{qos = 1, topic = <<"t3">>}, Q2),
  104. ?assertEqual(3, ?Q:len(Q3)),
  105. {_, Q4} = ?Q:in(#message{qos = 1, topic = <<"t2">>}, Q3),
  106. ?assertEqual(4, ?Q:len(Q4)),
  107. {_, Q5} = ?Q:in(#message{qos = 1, topic = <<"t2">>}, Q4),
  108. ?assertEqual(5, ?Q:len(Q5)),
  109. {_, Q6} = ?Q:in(#message{qos = 1, topic = <<"t2">>}, Q5),
  110. ?assertEqual(5, ?Q:len(Q6)),
  111. {{value, Msg}, Q7} = ?Q:out(Q6),
  112. ?assertEqual(4, ?Q:len(Q7)),
  113. ?assertEqual(<<"t3">>, Msg#message.topic).
  114. t_infinity_priority_mqueue(_) ->
  115. Opts = #{max_len => 0,
  116. priorities =>
  117. #{<<"t">> => 1,
  118. <<"t1">> => 2
  119. },
  120. store_qos0 => false},
  121. Q = ?Q:init(Opts),
  122. ?assertEqual(0, ?Q:max_len(Q)),
  123. Qx = lists:foldl(fun(I, AccQ) ->
  124. {undefined, AccQ1} = ?Q:in(#message{topic = <<"t1">>, qos = 1, payload = iolist_to_binary([I])}, AccQ),
  125. {undefined, AccQ2} = ?Q:in(#message{topic = <<"t">>, qos = 1, payload = iolist_to_binary([I])}, AccQ1),
  126. AccQ2
  127. end, Q, lists:seq(1, 255)),
  128. ?assertEqual(510, ?Q:len(Qx)),
  129. ?assertEqual([{len, 510}, {max_len, 0}, {dropped, 0}], ?Q:stats(Qx)).
  130. %%TODO: fixme later
  131. t_length_priority_mqueue(_) ->
  132. Opts = #{max_len => 2,
  133. store_qos0 => false
  134. },
  135. Q = ?Q:init(Opts),
  136. 2 = ?Q:max_len(Q),
  137. {_, Q1} = ?Q:in(#message{topic = <<"x">>, qos = 1, payload = <<1>>}, Q),
  138. {_, Q2} = ?Q:in(#message{topic = <<"x">>, qos = 1, payload = <<2>>}, Q1),
  139. {_, Q3} = ?Q:in(#message{topic = <<"y">>, qos = 1, payload = <<3>>}, Q2),
  140. {_, Q4} = ?Q:in(#message{topic = <<"y">>, qos = 1, payload = <<4>>}, Q3),
  141. ?assertEqual(2, ?Q:len(Q4)),
  142. {{value, _Val}, Q5} = ?Q:out(Q4),
  143. ?assertEqual(1, ?Q:len(Q5)).