Fix reload zone fail bug (#2857)

src/emqx_mqtt_caps.erl

@@ -126,31 +126,13 @@ default_caps() ->
     ?DEFAULT_CAPS.
 
 get_caps(Zone, publish) ->
-    with_env(Zone, '$mqtt_pub_caps',
+    filter_caps(?PUBCAP_KEYS, get_caps(Zone));
-             fun() ->
-                 filter_caps(?PUBCAP_KEYS, get_caps(Zone))
-             end);
 
 get_caps(Zone, subscribe) ->
-    with_env(Zone, '$mqtt_sub_caps',
+    filter_caps(?SUBCAP_KEYS, get_caps(Zone)).
-             fun() ->
-                 filter_caps(?SUBCAP_KEYS, get_caps(Zone))
-             end).
 
 get_caps(Zone) ->
-    with_env(Zone, '$mqtt_caps',
+    maps:from_list([{Cap, emqx_zone:get_env(Zone, Cap, Def)} || {Cap, Def} <- ?DEFAULT_CAPS]).
-             fun() ->
-                 maps:from_list([{Cap, emqx_zone:get_env(Zone, Cap, Def)}
-                                 || {Cap, Def} <- ?DEFAULT_CAPS])
-             end).
 
 filter_caps(Keys, Caps) ->
     maps:filter(fun(Key, _Val) -> lists:member(Key, Keys) end, Caps).
-
-with_env(Zone, Key, InitFun) ->
-    case emqx_zone:get_env(Zone, Key) of
-        undefined -> Caps = InitFun(),
-                     ok = emqx_zone:set_env(Zone, Key, Caps),
-                     Caps;
-        ZoneCaps  -> ZoneCaps
-    end.

test/emqx_client_SUITE.erl

@@ -70,6 +70,7 @@ receive_messages(Count, Msgs) ->
 basic_test(_Config) ->
     Topic = nth(1, ?TOPICS),
     ct:print("Basic test starting"),
+    init_caps(),
     {ok, C} = emqx_client:start_link(),
     {ok, _} = emqx_client:connect(C),
     {ok, _, [1]} = emqx_client:subscribe(C, Topic, qos1),
@@ -81,6 +82,7 @@ basic_test(_Config) ->
     ok = emqx_client:disconnect(C).
 
 will_message_test(_Config) ->
+    init_caps(),
     {ok, C1} = emqx_client:start_link([{clean_start, true},
                                           {will_topic, nth(3, ?TOPICS)},
                                           {will_payload, <<"client disconnected">>},
@@ -99,10 +101,10 @@ will_message_test(_Config) ->
     ct:print("Will message test succeeded").
 
 offline_message_queueing_test(_) ->
+    init_caps(),
     {ok, C1} = emqx_client:start_link([{clean_start, false},
                                        {client_id, <<"c1">>}]),
     {ok, _} = emqx_client:connect(C1),
-
     {ok, _, [2]} = emqx_client:subscribe(C1, nth(6, ?WILD_TOPICS), 2),
     ok = emqx_client:disconnect(C1),
     {ok, C2} = emqx_client:start_link([{clean_start, true},
@@ -123,6 +125,7 @@ offline_message_queueing_test(_) ->
     ?assertEqual(3, length(receive_messages(3))).
 
 overlapping_subscriptions_test(_) ->
+    init_caps(),
     {ok, C} = emqx_client:start_link([]),
     {ok, _} = emqx_client:connect(C),
 
@@ -163,6 +166,7 @@ overlapping_subscriptions_test(_) ->
 
 redelivery_on_reconnect_test(_) ->
     ct:print("Redelivery on reconnect test starting"),
+    init_caps(),
     {ok, C1} = emqx_client:start_link([{clean_start, false},
                                        {client_id, <<"c">>}]),
     {ok, _} = emqx_client:connect(C1),
@@ -194,6 +198,7 @@ redelivery_on_reconnect_test(_) ->
 
 dollar_topics_test(_) ->
     ct:print("$ topics test starting"),
+    init_caps(),
     {ok, C} = emqx_client:start_link([{clean_start, true},
                                       {keepalive, 0}]),
     {ok, _} = emqx_client:connect(C),
@@ -205,3 +210,13 @@ dollar_topics_test(_) ->
     ?assertEqual(0, length(receive_messages(1))),
     ok = emqx_client:disconnect(C),
     ct:print("$ topics test succeeded").
+
+init_caps() ->
+    Caps = #{max_qos_allowed => 2,
+             max_topic_levels => 0,
+             mqtt_shared_subscription => true,
+             mqtt_wildcard_subscription => true,
+             max_topic_alias => 0,
+             mqtt_retain_available => true},
+    [emqx_zone:set_env(external, Key, Val) ||{Key, Val} <- maps:to_list(Caps)],
+    timer:sleep(100).

test/emqx_mqtt_caps_SUITE.erl

@@ -27,38 +27,19 @@ all() -> [t_get_set_caps, t_check_pub, t_check_sub].
 
 t_get_set_caps(_) ->
     {ok, _} = emqx_zone:start_link(),
-    Caps = #{
+
-        max_packet_size => ?MAX_PACKET_SIZE,
+    PubCaps = emqx_mqtt_caps:get_caps(external, publish),
-        max_clientid_len => ?MAX_CLIENTID_LEN,
-        max_topic_alias => 0,
-        max_topic_levels => 0,
-        max_qos_allowed => ?QOS_2,
-        mqtt_retain_available => true,
-        mqtt_shared_subscription => true,
-        mqtt_wildcard_subscription => true
-    },
-    Caps2 = Caps#{max_packet_size => 1048576},
-    case emqx_mqtt_caps:get_caps(zone) of
-        Caps -> ok;
-        Caps2 -> ok
-    end,
-    PubCaps = #{
-        max_qos_allowed => ?QOS_2,
-        mqtt_retain_available => true,
-        max_topic_alias => 0
-    },
-    PubCaps = emqx_mqtt_caps:get_caps(zone, publish),
     NewPubCaps = PubCaps#{max_qos_allowed => ?QOS_1},
-    emqx_zone:set_env(zone, '$mqtt_pub_caps', NewPubCaps),
+    [emqx_zone:set_env(external, Key, Val) ||{Key, Val} <- maps:to_list(NewPubCaps)],
     timer:sleep(100),
-    NewPubCaps = emqx_mqtt_caps:get_caps(zone, publish),
+    NewPubCaps = emqx_mqtt_caps:get_caps(external, publish),
-    SubCaps = #{
+
-        max_topic_levels => 0,
+    SubCaps = emqx_mqtt_caps:get_caps(external, subscribe),
-        max_qos_allowed => ?QOS_2,
+    NewSubCaps = SubCaps#{max_topic_levels => 2},
-        mqtt_shared_subscription => true,
+    [emqx_zone:set_env(external, Key, Val) ||{Key, Val} <- maps:to_list(NewSubCaps)],
-        mqtt_wildcard_subscription => true
+    timer:sleep(100),
-    },
+    NewSubCaps = emqx_mqtt_caps:get_caps(external, subscribe),
-    SubCaps = emqx_mqtt_caps:get_caps(zone, subscribe),
+
     emqx_zone:stop().
 
 t_check_pub(_) ->
@@ -68,35 +49,34 @@ t_check_pub(_) ->
         mqtt_retain_available => false,
         max_topic_alias => 4
     },
-    emqx_zone:set_env(zone, '$mqtt_pub_caps', PubCaps),
+    [emqx_zone:set_env(external, Key, Val) ||{Key, Val} <- maps:to_list(PubCaps)],
     timer:sleep(100),
-    ct:log("~p", [emqx_mqtt_caps:get_caps(zone, publish)]),
+    ct:log("~p", [emqx_mqtt_caps:get_caps(external, publish)]),
     BadPubProps1 = #{
         qos => ?QOS_2,
         retain => false
     },
-    {error, ?RC_QOS_NOT_SUPPORTED} = emqx_mqtt_caps:check_pub(zone, BadPubProps1),
+    {error, ?RC_QOS_NOT_SUPPORTED} = emqx_mqtt_caps:check_pub(external, BadPubProps1),
     BadPubProps2 = #{
         qos => ?QOS_1,
         retain => true
     },
-    {error, ?RC_RETAIN_NOT_SUPPORTED} = emqx_mqtt_caps:check_pub(zone, BadPubProps2),
+    {error, ?RC_RETAIN_NOT_SUPPORTED} = emqx_mqtt_caps:check_pub(external, BadPubProps2),
     BadPubProps3 = #{
         qos => ?QOS_1,
         retain => false,
         topic_alias => 5
     },
-    {error, ?RC_TOPIC_ALIAS_INVALID} = emqx_mqtt_caps:check_pub(zone, BadPubProps3),
+    {error, ?RC_TOPIC_ALIAS_INVALID} = emqx_mqtt_caps:check_pub(external, BadPubProps3),
     PubProps = #{
         qos => ?QOS_1,
         retain => false
     },
-    ok = emqx_mqtt_caps:check_pub(zone, PubProps),
+    ok = emqx_mqtt_caps:check_pub(external, PubProps),
     emqx_zone:stop().
 
 t_check_sub(_) ->
     {ok, _} = emqx_zone:start_link(),
-
     Opts = #{qos => ?QOS_2, share => true, rc => 0},
     Caps = #{
         max_topic_levels => 0,
@@ -104,6 +84,8 @@ t_check_sub(_) ->
         mqtt_shared_subscription => true,
         mqtt_wildcard_subscription => true
     },
+    [emqx_zone:set_env(external, Key, Val) ||{Key, Val} <- maps:to_list(Caps)],
+    timer:sleep(100),
 
     ok = do_check_sub([{<<"client/stat">>, Opts}], [{<<"client/stat">>, Opts}]),
     ok = do_check_sub(Caps#{max_qos_allowed => ?QOS_1}, [{<<"client/stat">>, Opts}], [{<<"client/stat">>, Opts#{qos => ?QOS_1}}]),
@@ -122,10 +104,10 @@ t_check_sub(_) ->
 
 
 do_check_sub(TopicFilters, Topics) ->
-    {ok, Topics} = emqx_mqtt_caps:check_sub(zone, TopicFilters),
+    {ok, Topics} = emqx_mqtt_caps:check_sub(external, TopicFilters),
     ok.
 do_check_sub(Caps, TopicFilters, Topics) ->
-    emqx_zone:set_env(zone, '$mqtt_sub_caps', Caps),
+    [emqx_zone:set_env(external, Key, Val) ||{Key, Val} <- maps:to_list(Caps)],
     timer:sleep(100),
-    {_, Topics} = emqx_mqtt_caps:check_sub(zone, TopicFilters),
+    {_, Topics} = emqx_mqtt_caps:check_sub(external, TopicFilters),
     ok.

test/emqx_protocol_SUITE.erl

@@ -56,7 +56,7 @@ groups() ->
 init_per_suite(Config) ->
     emqx_ct_helpers:start_apps([], fun set_special_configs/1),
     MqttCaps = maps:from_list(emqx_mqtt_caps:default_caps()),
-    emqx_zone:set_env(external, '$mqtt_caps', MqttCaps#{max_topic_alias => 20}),
+    [emqx_zone:set_env(external, Key, Val) ||{Key, Val} <- maps:to_list(MqttCaps#{max_topic_alias => 20})],
     Config.
 
 end_per_suite(_Config) ->

test/emqx_session_SUITE.erl

@@ -89,7 +89,7 @@ message_expiry_interval_exipred(ClientA, QoS) ->
     ct:pal("~p ~p", [?FUNCTION_NAME, QoS]),
     %% publish to t/a and waiting for the message expired
     emqx_client:publish(ClientA, <<"t/a">>, #{'Message-Expiry-Interval' => 1}, <<"this will be purged in 1s">>, [{qos, QoS}]),
-    ct:sleep(1000),
+    ct:sleep(2000),
 
     %% resume the session for client-b
     {ok, ClientB1} = emqx_client:start_link([{proto_ver,v5}, {client_id, <<"client-b">>}, {clean_start, false},{properties, #{'Session-Expiry-Interval' => 360}}]),