Update the test cases for emqttd_router

test/emqttd_SUITE.erl

@@ -240,8 +240,10 @@ t_local_subscribe(_) ->
     emqttd:subscribe("$local/topic2", <<"x">>, [{qos, 2}]),
     timer:sleep(10),
     ?assertEqual([self()], emqttd:subscribers("$local/topic0")),
-    ?assertEqual([<<"x">>], emqttd:subscribers("$local/topic1")),
+    ?assertEqual([{<<"x">>, self()}], emqttd:subscribers("$local/topic1")),
-    ?assertEqual([{<<"$local/topic1">>,<<"x">>,[]},{<<"$local/topic2">>,<<"x">>,[{qos,2}]}], emqttd:subscriptions(<<"x">>)),
+    ?assertEqual([{{<<"x">>, self()}, <<"$local/topic1">>, []},
+                  {{<<"x">>, self()}, <<"$local/topic2">>, [{qos,2}]}],
+                 emqttd:subscriptions(<<"x">>)),
     
     ?assertEqual(ok, emqttd:unsubscribe("$local/topic0")),
     ?assertMatch({error, {subscription_not_found, _}}, emqttd:unsubscribe("$local/topic0")),
@@ -256,9 +258,9 @@ t_shared_subscribe(_) ->
     emqttd:subscribe("$queue/topic3"),
     timer:sleep(10),
     ?assertEqual([self()], emqttd:subscribers(<<"$local/$share/group1/topic1">>)),
-    ?assertEqual([{<<"$local/$share/group1/topic1">>, self(), []},
+    ?assertEqual([{self(), <<"$local/$share/group1/topic1">>, []},
-                  {<<"$queue/topic3">>, self(), []},
+                  {self(), <<"$queue/topic3">>, []},
-                  {<<"$share/group2/topic2">>, self(), []}],
+                  {self(), <<"$share/group2/topic2">>, []}],
                  lists:sort(emqttd:subscriptions(self()))),
     emqttd:unsubscribe("$local/$share/group1/topic1"),
     emqttd:unsubscribe("$share/group2/topic2"),
@@ -298,7 +300,7 @@ router_add_del(_) ->
     %% Add
     emqttd_router:add_route(<<"#">>),
     emqttd_router:add_route(<<"a/b/c">>),
-    emqttd_router:add_route(<<"+/#">>, node()),
+    emqttd_router:add_route(<<"+/#">>),
     Routes = [R1, R2 | _] = [
             #mqtt_route{topic = <<"#">>,     node = node()},
             #mqtt_route{topic = <<"+/#">>,   node = node()},
@@ -306,7 +308,7 @@ router_add_del(_) ->
     Routes = lists:sort(emqttd_router:match(<<"a/b/c">>)),
 
     %% Batch Add
-    emqttd_router:add_routes(Routes),
+    lists:foreach(fun(R) -> emqttd_router:add_route(R) end, Routes),
     Routes = lists:sort(emqttd_router:match(<<"a/b/c">>)),
 
     %% Del
@@ -317,7 +319,8 @@ router_add_del(_) ->
     %% Batch Del
     R3 = #mqtt_route{topic = <<"#">>, node = 'a@127.0.0.1'},
     emqttd_router:add_route(R3),
-    emqttd_router:del_routes([R1, R2]),
+    emqttd_router:del_route(R1),
+    emqttd_router:del_route(R2),
     emqttd_router:del_route(R3),
     [] = lists:sort(emqttd_router:match(<<"a/b/c">>)).
 
@@ -325,7 +328,7 @@ router_print(_) ->
     Routes = [#mqtt_route{topic = <<"a/b/c">>, node = node()},
               #mqtt_route{topic = <<"#">>,     node = node()},
               #mqtt_route{topic = <<"+/#">>,   node = node()}],
-    emqttd_router:add_routes(Routes),
+    lists:foreach(fun(R) -> emqttd_router:add_route(R) end, Routes),
     emqttd_router:print(<<"a/b/c">>).
 
 router_unused(_) ->
@@ -591,7 +594,7 @@ conflict_listeners(_) ->
               end, esockd:listeners()),
     L = proplists:get_value("mqtt:tcp:0.0.0.0:1883", Listeners),
     ?assertEqual(1, proplists:get_value(current_clients, L)),
-    ?assertEqual(1, proplists:get_value(conflict, L)),
+    ?assertEqual(1, proplists:get_value(conflict, proplists:get_value(shutdown_count, L))),
     emqttc:disconnect(C2).
 
 cli_vm(_) ->

test/emqttd_router_SUITE.erl

@@ -0,0 +1,132 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2013-2017 EMQ Enterprise, Inc. (http://emqtt.io)
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+
+-module(emqttd_router_SUITE).
+
+-compile(export_all).
+
+-include("emqttd.hrl").
+
+-include_lib("eunit/include/eunit.hrl").
+
+-define(R, emqttd_router).
+
+all() ->
+    [{group, route},
+     {group, local_route}].
+
+groups() ->
+    [{route, [sequence],
+      [t_get_topics,
+       t_add_del_route,
+       t_match_route,
+       t_print,
+       t_has_route]},
+     {local_route, [sequence],
+      [t_get_local_topics,
+       t_add_del_local_route,
+       t_match_local_route]}].
+
+init_per_suite(Config) ->
+    ekka:start(),
+    ekka_mnesia:ensure_started(),
+    {ok, _R} = emqttd_router:start(),
+    Config.
+
+end_per_suite(_Config) ->
+    emqttd_router:stop(),
+    ekka:stop(),
+    ekka_mnesia:ensure_stopped(),
+    ekka_mnesia:delete_schema().
+
+init_per_testcase(_TestCase, Config) ->
+    Config.
+
+end_per_testcase(_TestCase, _Config) ->
+    clear_tables().
+
+t_get_topics(_) ->
+    ?R:add_route(<<"a/b/c">>),
+    ?R:add_route(<<"a/b/c">>),
+    ?R:add_route(<<"a/+/b">>),
+    ?assertEqual([<<"a/+/b">>, <<"a/b/c">>], lists:sort(?R:topics())),
+    ?R:del_route(<<"a/b/c">>),
+    ?R:del_route(<<"a/+/b">>),
+    ?assertEqual([], lists:sort(?R:topics())).
+
+t_add_del_route(_) ->
+    %%Node = node(),
+    ?R:add_route(<<"a/b/c">>),
+    ?R:add_route(<<"a/+/b">>),
+    ?R:del_route(<<"a/b/c">>),
+    ?R:del_route(<<"a/+/b">>).
+
+t_match_route(_) ->
+    Node = node(),
+    ?R:add_route(<<"a/b/c">>),
+    ?R:add_route(<<"a/+/c">>),
+    ?R:add_route(<<"a/b/#">>),
+    ?R:add_route(<<"#">>),
+    ?assertEqual([#mqtt_route{topic = <<"#">>, node = Node},
+                  #mqtt_route{topic = <<"a/+/c">>, node = Node},
+                  #mqtt_route{topic = <<"a/b/#">>, node = Node},
+                  #mqtt_route{topic = <<"a/b/c">>, node = Node}],
+                 lists:sort(?R:match(<<"a/b/c">>))).
+
+t_print(_) ->
+    ?R:add_route(<<"topic">>),
+    ?R:add_route(<<"topic/#">>),
+    ?R:print(<<"topic">>).
+
+t_has_route(_) ->
+    ?R:add_route(<<"devices/+/messages">>),
+    ?assert(?R:has_route(<<"devices/+/messages">>)).
+
+t_get_local_topics(_) ->
+    ?R:add_local_route(<<"a/b/c">>),
+    ?R:add_local_route(<<"x/+/y">>),
+    ?R:add_local_route(<<"z/#">>),
+    ?assertEqual([<<"z/#">>, <<"x/+/y">>, <<"a/b/c">>], ?R:local_topics()),
+    ?R:del_local_route(<<"x/+/y">>),
+    ?R:del_local_route(<<"z/#">>),
+    ?assertEqual([<<"a/b/c">>], ?R:local_topics()).
+
+t_add_del_local_route(_) ->
+    Node = node(),
+    ?R:add_local_route(<<"a/b/c">>),
+    ?R:add_local_route(<<"x/+/y">>),
+    ?R:add_local_route(<<"z/#">>),
+    ?assertEqual([{<<"a/b/c">>, Node},
+                  {<<"x/+/y">>, Node},
+                  {<<"z/#">>, Node}],
+                 lists:sort(?R:get_local_routes())),
+    ?R:del_local_route(<<"x/+/y">>),
+    ?R:del_local_route(<<"z/#">>),
+    ?assertEqual([{<<"a/b/c">>, Node}], lists:sort(?R:get_local_routes())).
+
+t_match_local_route(_) ->
+    ?R:add_local_route(<<"$SYS/#">>),
+    ?R:add_local_route(<<"a/b/c">>),
+    ?R:add_local_route(<<"a/+/c">>),
+    ?R:add_local_route(<<"a/b/#">>),
+    ?R:add_local_route(<<"#">>),
+    Matched = [Topic || #mqtt_route{topic = {local, Topic}} <- ?R:match_local(<<"a/b/c">>)],
+    ?assertEqual([<<"#">>, <<"a/+/c">>, <<"a/b/#">>, <<"a/b/c">>], lists:sort(Matched)).
+
+clear_tables() ->
+    ?R:clean_local_routes(),
+    lists:foreach(fun mnesia:clear_table/1, [mqtt_route, mqtt_trie, mqtt_trie_node]).
+