priority mqueue test

test/emqttd_mqueue_tests.erl

@@ -24,7 +24,7 @@
 
 -include("emqttd.hrl").
 
--define(QM, emqttd_mqueue).
+-define(Q, emqttd_mqueue).
 
 -ifdef(TEST).
 
@@ -33,39 +33,77 @@
 in_test() ->
     Opts = [{max_length, 5},
             {queue_qos0, true}],
-    Q = ?QM:new(<<"testQ">>, Opts, alarm_fun()),
-    ?assertEqual(true, ?QM:is_empty(Q)),
-    Q1 = ?QM:in(#mqtt_message{}, Q),
-    ?assertEqual(1, ?QM:len(Q1)),
-    Q2 = ?QM:in(#mqtt_message{qos = 1}, Q1),
-    ?assertEqual(2, ?QM:len(Q2)),
-    Q3 = ?QM:in(#mqtt_message{qos = 2}, Q2),
-    Q4 = ?QM:in(#mqtt_message{}, Q3),
-    Q5 = ?QM:in(#mqtt_message{}, Q4),
-    ?assertEqual(5, ?QM:len(Q5)).
+    Q = ?Q:new(<<"testQ">>, Opts, alarm_fun()),
+    ?assertEqual(true, ?Q:is_empty(Q)),
+    Q1 = ?Q:in(#mqtt_message{}, Q),
+    ?assertEqual(1, ?Q:len(Q1)),
+    Q2 = ?Q:in(#mqtt_message{qos = 1}, Q1),
+    ?assertEqual(2, ?Q:len(Q2)),
+    Q3 = ?Q:in(#mqtt_message{qos = 2}, Q2),
+    Q4 = ?Q:in(#mqtt_message{}, Q3),
+    Q5 = ?Q:in(#mqtt_message{}, Q4),
+    ?assertEqual(5, ?Q:len(Q5)).
     
 in_qos0_test() ->
     Opts = [{max_length, 5},
             {queue_qos0, false}],
-    Q = ?QM:new(<<"testQ">>, Opts, alarm_fun()),
-    Q1 = ?QM:in(#mqtt_message{}, Q),
-    ?assertEqual(true, ?QM:is_empty(Q1)),
-    Q2 = ?QM:in(#mqtt_message{qos = 0}, Q1),
-    ?assertEqual(true, ?QM:is_empty(Q2)).
+    Q = ?Q:new(<<"testQ">>, Opts, alarm_fun()),
+    Q1 = ?Q:in(#mqtt_message{}, Q),
+    ?assertEqual(true, ?Q:is_empty(Q1)),
+    Q2 = ?Q:in(#mqtt_message{qos = 0}, Q1),
+    ?assertEqual(true, ?Q:is_empty(Q2)).
 
 out_test() ->
     Opts = [{max_length, 5},
             {queue_qos0, true}],
-    Q = ?QM:new(<<"testQ">>, Opts, alarm_fun()),
-    ?assertMatch({empty, Q}, ?QM:out(Q)),
-    Q1 = ?QM:in(#mqtt_message{}, Q),
-    {Value, Q2} = ?QM:out(Q1),
-    ?assertEqual(0, ?QM:len(Q2)),
+    Q = ?Q:new(<<"testQ">>, Opts, alarm_fun()),
+    ?assertMatch({empty, Q}, ?Q:out(Q)),
+    Q1 = ?Q:in(#mqtt_message{}, Q),
+    {Value, Q2} = ?Q:out(Q1),
+    ?assertEqual(0, ?Q:len(Q2)),
     ?assertMatch({value, #mqtt_message{}}, Value).
+
+simple_mqueue_test() ->
+    Opts = [{type, simple},
+            {max_length, 3},
+            {low_watermark, 0.2},
+            {high_watermark, 0.6},
+            {queue_qos0, false}],
+    Q = ?Q:new("simple_queue", Opts, alarm_fun()),
+
+    ?assert(?Q:is_empty(Q)),
+    Q1 = ?Q:in(#mqtt_message{qos = 1, payload = <<"1">>}, Q),
+    Q2 = ?Q:in(#mqtt_message{qos = 1, payload = <<"2">>}, Q1),
+    Q3 = ?Q:in(#mqtt_message{qos = 1, payload = <<"3">>}, Q2),
+    Q4 = ?Q:in(#mqtt_message{qos = 1, payload = <<"4">>}, Q3),
+    ?assertEqual(3, ?Q:len(Q4)),
+    {{value, Msg}, Q5} = ?Q:out(Q4),
+    ?assertMatch(<<"2">>, Msg#mqtt_message.payload).
+
+priority_mqueue_test() ->
+    Opts = [{type, priority},
+            {priority, [{<<"t">>, 10}]},
+            {max_length, 3},
+            {low_watermark, 0.2},
+            {high_watermark, 0.6},
+            {queue_qos0, false}],
+    Q = ?Q:new("priority_queue", Opts, alarm_fun()),
+
+    ?assert(?Q:is_empty(Q)),
+    Q1 = ?Q:in(#mqtt_message{qos = 1, topic = <<"t1">>}, Q),
+    Q2 = ?Q:in(#mqtt_message{qos = 1, topic = <<"t">>}, Q1),
+    Q3 = ?Q:in(#mqtt_message{qos = 1, topic = <<"t2">>}, Q2),
+    ?assertEqual(3, ?Q:len(Q3)),
+    Q4 = ?Q:in(#mqtt_message{qos = 1, topic = <<"t1">>}, Q3),
+    ?assertEqual(4, ?Q:len(Q4)),
+    Q5 = ?Q:in(#mqtt_message{qos = 1, topic = <<"t1">>}, Q4),
+    ?assertEqual(5, ?Q:len(Q5)),
+    Q6 = ?Q:in(#mqtt_message{qos = 1, topic = <<"t1">>}, Q5),
+    ?assertEqual(5, ?Q:len(Q6)),
+    {{value, Msg}, Q7} = ?Q:out(Q6),
+    ?assertMatch(<<"t">>, Msg#mqtt_message.topic).
  
-alarm_fun() ->
-   fun(_, _) -> alarm_fun() end.
+alarm_fun() -> fun(_, _) -> alarm_fun() end.
 
 -endif.
 
-