Merge branch 'emqx:master' into master

README-CN.md

@@ -49,7 +49,7 @@ docker run -d --name emqx -p 1883:1883 -p 8081:8081 -p 8083:8083 -p 8883:8883 -p
 git clone https://github.com/emqx/emqx.git
 cd emqx
 make
-_build/emqx/rel/emqx/bin console
+_build/emqx/rel/emqx/bin/emqx console
 ```
 
 对于 4.3 之前的版本，通过另外一个仓库构建：

README-RU.md

@@ -50,7 +50,7 @@ docker run -d --name emqx -p 1883:1883 -p 8081:8081 -p 8083:8083 -p 8883:8883 -p
 git clone https://github.com/emqx/emqx.git
 cd emqx
 make
-_build/emqx/rel/emqx/bin console
+_build/emqx/rel/emqx/bin/emqx console
 ```
 
 Более ранние релизы могут быть собраны с помощью другого репозитория:

README.md

@@ -49,7 +49,7 @@ For 4.3 and later versions.
 git clone https://github.com/emqx/emqx.git
 cd emqx
 make
-_build/emqx/rel/emqx/bin console
+_build/emqx/rel/emqx/bin/emqx console
 ```
 
 For earlier versions, release has to be built from another repo.

apps/emqx/src/emqx.erl

@@ -119,17 +119,17 @@ is_running(Node) ->
 %% PubSub API
 %%--------------------------------------------------------------------
 
--spec(subscribe(emqx_topic:topic() | string()) -> ok).
+-spec(subscribe(emqx_types:topic() | string()) -> ok).
 subscribe(Topic) ->
     emqx_broker:subscribe(iolist_to_binary(Topic)).
 
--spec(subscribe(emqx_topic:topic() | string(), emqx_types:subid() | emqx_types:subopts()) -> ok).
+-spec(subscribe(emqx_types:topic() | string(), emqx_types:subid() | emqx_types:subopts()) -> ok).
 subscribe(Topic, SubId) when is_atom(SubId); is_binary(SubId)->
     emqx_broker:subscribe(iolist_to_binary(Topic), SubId);
 subscribe(Topic, SubOpts) when is_map(SubOpts) ->
     emqx_broker:subscribe(iolist_to_binary(Topic), SubOpts).
 
--spec(subscribe(emqx_topic:topic() | string(),
+-spec(subscribe(emqx_types:topic() | string(),
                 emqx_types:subid() | pid(), emqx_types:subopts()) -> ok).
 subscribe(Topic, SubId, SubOpts) when (is_atom(SubId) orelse is_binary(SubId)), is_map(SubOpts) ->
     emqx_broker:subscribe(iolist_to_binary(Topic), SubId, SubOpts).
@@ -138,7 +138,7 @@ subscribe(Topic, SubId, SubOpts) when (is_atom(SubId) orelse is_binary(SubId)),
 publish(Msg) ->
     emqx_broker:publish(Msg).
 
--spec(unsubscribe(emqx_topic:topic() | string()) -> ok).
+-spec(unsubscribe(emqx_types:topic() | string()) -> ok).
 unsubscribe(Topic) ->
     emqx_broker:unsubscribe(iolist_to_binary(Topic)).
 
@@ -146,18 +146,18 @@ unsubscribe(Topic) ->
 %% PubSub management API
 %%--------------------------------------------------------------------
 
--spec(topics() -> list(emqx_topic:topic())).
+-spec(topics() -> list(emqx_types:topic())).
 topics() -> emqx_router:topics().
 
--spec(subscribers(emqx_topic:topic() | string()) -> [pid()]).
+-spec(subscribers(emqx_types:topic() | string()) -> [pid()]).
 subscribers(Topic) ->
     emqx_broker:subscribers(iolist_to_binary(Topic)).
 
--spec(subscriptions(pid()) -> [{emqx_topic:topic(), emqx_types:subopts()}]).
+-spec(subscriptions(pid()) -> [{emqx_types:topic(), emqx_types:subopts()}]).
 subscriptions(SubPid) when is_pid(SubPid) ->
     emqx_broker:subscriptions(SubPid).
 
--spec(subscribed(pid() | emqx_types:subid(), emqx_topic:topic() | string()) -> boolean()).
+-spec(subscribed(pid() | emqx_types:subid(), emqx_types:topic() | string()) -> boolean()).
 subscribed(SubPid, Topic) when is_pid(SubPid) ->
     emqx_broker:subscribed(SubPid, iolist_to_binary(Topic));
 subscribed(SubId, Topic) when is_atom(SubId); is_binary(SubId) ->

apps/emqx/src/emqx_authz_cache.erl

@@ -68,7 +68,7 @@ list_authz_cache() ->
     map_authz_cache(fun(Cache) -> Cache end).
 
 %% We'll cleanup the cache before replacing an expired authz.
--spec get_authz_cache(emqx_types:pubsub(), emqx_topic:topic()) ->
+-spec get_authz_cache(emqx_types:pubsub(), emqx_types:topic()) ->
     authz_result() | not_found.
 get_authz_cache(PubSub, Topic) ->
     case erlang:get(cache_k(PubSub, Topic)) of
@@ -85,7 +85,7 @@ get_authz_cache(PubSub, Topic) ->
 
 %% If the cache get full, and also the latest one
 %%   is expired, then delete all the cache entries
--spec put_authz_cache(emqx_types:pubsub(), emqx_topic:topic(), authz_result())
+-spec put_authz_cache(emqx_types:pubsub(), emqx_types:topic(), authz_result())
     -> ok.
 put_authz_cache(PubSub, Topic, AuthzResult) ->
     MaxSize = get_cache_max_size(), true = (MaxSize =/= 0),

apps/emqx/src/emqx_broker.erl

@@ -112,17 +112,17 @@ create_tabs() ->
 %% Subscribe API
 %%------------------------------------------------------------------------------
 
--spec(subscribe(emqx_topic:topic()) -> ok).
+-spec(subscribe(emqx_types:topic()) -> ok).
 subscribe(Topic) when is_binary(Topic) ->
     subscribe(Topic, undefined).
 
--spec(subscribe(emqx_topic:topic(), emqx_types:subid() | emqx_types:subopts()) -> ok).
+-spec(subscribe(emqx_types:topic(), emqx_types:subid() | emqx_types:subopts()) -> ok).
 subscribe(Topic, SubId) when is_binary(Topic), ?is_subid(SubId) ->
     subscribe(Topic, SubId, ?DEFAULT_SUBOPTS);
 subscribe(Topic, SubOpts) when is_binary(Topic), is_map(SubOpts) ->
     subscribe(Topic, undefined, SubOpts).
 
--spec(subscribe(emqx_topic:topic(), emqx_types:subid(), emqx_types:subopts()) -> ok).
+-spec(subscribe(emqx_types:topic(), emqx_types:subid(), emqx_types:subopts()) -> ok).
 subscribe(Topic, SubId, SubOpts0) when is_binary(Topic), ?is_subid(SubId), is_map(SubOpts0) ->
     SubOpts = maps:merge(?DEFAULT_SUBOPTS, SubOpts0),
     case ets:member(?SUBOPTION, {SubPid = self(), Topic}) of
@@ -165,7 +165,7 @@ do_subscribe(Group, Topic, SubPid, SubOpts) ->
 %% Unsubscribe API
 %%--------------------------------------------------------------------
 
--spec(unsubscribe(emqx_topic:topic()) -> ok).
+-spec(unsubscribe(emqx_types:topic()) -> ok).
 unsubscribe(Topic) when is_binary(Topic) ->
     SubPid = self(),
     case ets:lookup(?SUBOPTION, {SubPid, Topic}) of
@@ -279,7 +279,7 @@ forward(Node, To, Delivery, sync) ->
             emqx_metrics:inc('messages.forward'), Result
     end.
 
--spec(dispatch(emqx_topic:topic(), emqx_types:delivery()) -> emqx_types:deliver_result()).
+-spec(dispatch(emqx_types:topic(), emqx_types:delivery()) -> emqx_types:deliver_result()).
 dispatch(Topic, #delivery{message = Msg}) ->
     DispN = lists:foldl(
                 fun(Sub, N) ->
@@ -316,7 +316,7 @@ inc_dropped_cnt(Msg) ->
     end.
 
 -compile({inline, [subscribers/1]}).
--spec(subscribers(emqx_topic:topic() | {shard, emqx_topic:topic(), non_neg_integer()})
+-spec(subscribers(emqx_types:topic() | {shard, emqx_types:topic(), non_neg_integer()})
       -> [pid()]).
 subscribers(Topic) when is_binary(Topic) ->
     lookup_value(?SUBSCRIBER, Topic, []);
@@ -351,7 +351,7 @@ subscriber_down(SubPid) ->
 %%--------------------------------------------------------------------
 
 -spec(subscriptions(pid() | emqx_types:subid())
-      -> [{emqx_topic:topic(), emqx_types:subopts()}]).
+      -> [{emqx_types:topic(), emqx_types:subopts()}]).
 subscriptions(SubPid) when is_pid(SubPid) ->
     [{Topic, lookup_value(?SUBOPTION, {SubPid, Topic}, #{})}
       || Topic <- lookup_value(?SUBSCRIPTION, SubPid, [])];
@@ -362,14 +362,14 @@ subscriptions(SubId) ->
         undefined -> []
     end.
 
--spec(subscribed(pid() | emqx_types:subid(), emqx_topic:topic()) -> boolean()).
+-spec(subscribed(pid() | emqx_types:subid(), emqx_types:topic()) -> boolean()).
 subscribed(SubPid, Topic) when is_pid(SubPid) ->
     ets:member(?SUBOPTION, {SubPid, Topic});
 subscribed(SubId, Topic) when ?is_subid(SubId) ->
     SubPid = emqx_broker_helper:lookup_subpid(SubId),
     ets:member(?SUBOPTION, {SubPid, Topic}).
 
--spec(get_subopts(pid(), emqx_topic:topic()) -> maybe(emqx_types:subopts())).
+-spec(get_subopts(pid(), emqx_types:topic()) -> maybe(emqx_types:subopts())).
 get_subopts(SubPid, Topic) when is_pid(SubPid), is_binary(Topic) ->
     lookup_value(?SUBOPTION, {SubPid, Topic});
 get_subopts(SubId, Topic) when ?is_subid(SubId) ->
@@ -379,7 +379,7 @@ get_subopts(SubId, Topic) when ?is_subid(SubId) ->
         undefined -> undefined
     end.
 
--spec(set_subopts(emqx_topic:topic(), emqx_types:subopts()) -> boolean()).
+-spec(set_subopts(emqx_types:topic(), emqx_types:subopts()) -> boolean()).
 set_subopts(Topic, NewOpts) when is_binary(Topic), is_map(NewOpts) ->
     set_subopts(self(), Topic, NewOpts).
 
@@ -392,7 +392,7 @@ set_subopts(SubPid, Topic, NewOpts) ->
         [] -> false
     end.
 
--spec(topics() -> [emqx_topic:topic()]).
+-spec(topics() -> [emqx_types:topic()]).
 topics() ->
     emqx_router:topics().
 

apps/emqx/src/emqx_broker_helper.erl

@@ -78,7 +78,7 @@ lookup_subid(SubPid) when is_pid(SubPid) ->
 lookup_subpid(SubId) ->
     emqx_tables:lookup_value(?SUBID, SubId).
 
--spec(get_sub_shard(pid(), emqx_topic:topic()) -> non_neg_integer()).
+-spec(get_sub_shard(pid(), emqx_types:topic()) -> non_neg_integer()).
 get_sub_shard(SubPid, Topic) ->
     case create_seq(Topic) of
         Seq when Seq =< ?SHARD -> 0;
@@ -90,11 +90,11 @@ shards_num() ->
     %% Dynamic sharding later...
     ets:lookup_element(?HELPER, shards, 2).
 
--spec(create_seq(emqx_topic:topic()) -> emqx_sequence:seqid()).
+-spec(create_seq(emqx_types:topic()) -> emqx_sequence:seqid()).
 create_seq(Topic) ->
     emqx_sequence:nextval(?SUBSEQ, Topic).
 
--spec(reclaim_seq(emqx_topic:topic()) -> emqx_sequence:seqid()).
+-spec(reclaim_seq(emqx_types:topic()) -> emqx_sequence:seqid()).
 reclaim_seq(Topic) ->
     emqx_sequence:reclaim(?SUBSEQ, Topic).
 

apps/emqx/src/emqx_message.erl

@@ -86,19 +86,19 @@
 
 -elvis([{elvis_style, god_modules, disable}]).
 
--spec(make(emqx_topic:topic(), emqx_types:payload()) -> emqx_types:message()).
+-spec(make(emqx_types:topic(), emqx_types:payload()) -> emqx_types:message()).
 make(Topic, Payload) ->
     make(undefined, Topic, Payload).
 
 -spec(make(emqx_types:clientid(),
-           emqx_topic:topic(),
+           emqx_types:topic(),
            emqx_types:payload()) -> emqx_types:message()).
 make(From, Topic, Payload) ->
     make(From, ?QOS_0, Topic, Payload).
 
 -spec(make(emqx_types:clientid(),
            emqx_types:qos(),
-           emqx_topic:topic(),
+           emqx_types:topic(),
            emqx_types:payload()) -> emqx_types:message()).
 make(From, QoS, Topic, Payload) when ?QOS_0 =< QoS, QoS =< ?QOS_2 ->
     Now = erlang:system_time(millisecond),
@@ -112,7 +112,7 @@ make(From, QoS, Topic, Payload) when ?QOS_0 =< QoS, QoS =< ?QOS_2 ->
 
 -spec(make(emqx_types:clientid(),
            emqx_types:qos(),
-           emqx_topic:topic(),
+           emqx_types:topic(),
            emqx_types:payload(),
            emqx_types:flags(),
            emqx_types:headers()) -> emqx_types:message()).
@@ -133,7 +133,7 @@ make(From, QoS, Topic, Payload, Flags, Headers)
 -spec(make(MsgId :: binary(),
            emqx_types:clientid(),
            emqx_types:qos(),
-           emqx_topic:topic(),
+           emqx_types:topic(),
            emqx_types:payload(),
            emqx_types:flags(),
            emqx_types:headers()) -> emqx_types:message()).

apps/emqx/src/emqx_mqtt_caps.erl

@@ -67,7 +67,7 @@
 -spec(check_pub(emqx_types:zone(),
                 #{qos := emqx_types:qos(),
                   retain := boolean(),
-                  topic := emqx_topic:topic()})
+                  topic := emqx_types:topic()})
       -> ok_or_error(emqx_types:reason_code())).
 check_pub(Zone, Flags) when is_map(Flags) ->
     do_check_pub(case maps:take(topic, Flags) of

apps/emqx/src/emqx_mqueue.erl

@@ -71,7 +71,7 @@
 
 -export_type([mqueue/0, options/0]).
 
--type(topic() :: emqx_topic:topic()).
+-type(topic() :: emqx_types:topic()).
 -type(priority() :: infinity | integer()).
 -type(pq() :: emqx_pqueue:q()).
 -type(count() :: non_neg_integer()).

apps/emqx/src/emqx_router.erl

@@ -98,19 +98,19 @@ start_link(Pool, Id) ->
 %% Route APIs
 %%--------------------------------------------------------------------
 
--spec(add_route(emqx_topic:topic()) -> ok | {error, term()}).
+-spec(add_route(emqx_types:topic()) -> ok | {error, term()}).
 add_route(Topic) when is_binary(Topic) ->
     add_route(Topic, node()).
 
--spec(add_route(emqx_topic:topic(), dest()) -> ok | {error, term()}).
+-spec(add_route(emqx_types:topic(), dest()) -> ok | {error, term()}).
 add_route(Topic, Dest) when is_binary(Topic) ->
     call(pick(Topic), {add_route, Topic, Dest}).
 
--spec(do_add_route(emqx_topic:topic()) -> ok | {error, term()}).
+-spec(do_add_route(emqx_types:topic()) -> ok | {error, term()}).
 do_add_route(Topic) when is_binary(Topic) ->
     do_add_route(Topic, node()).
 
--spec(do_add_route(emqx_topic:topic(), dest()) -> ok | {error, term()}).
+-spec(do_add_route(emqx_types:topic(), dest()) -> ok | {error, term()}).
 do_add_route(Topic, Dest) when is_binary(Topic) ->
     Route = #route{topic = Topic, dest = Dest},
     case lists:member(Route, lookup_routes(Topic)) of
@@ -125,7 +125,7 @@ do_add_route(Topic, Dest) when is_binary(Topic) ->
     end.
 
 %% @doc Match routes
--spec(match_routes(emqx_topic:topic()) -> [emqx_types:route()]).
+-spec(match_routes(emqx_types:topic()) -> [emqx_types:route()]).
 match_routes(Topic) when is_binary(Topic) ->
     case match_trie(Topic) of
         [] -> lookup_routes(Topic);
@@ -140,27 +140,27 @@ match_trie(Topic) ->
         false -> emqx_trie:match(Topic)
     end.
 
--spec(lookup_routes(emqx_topic:topic()) -> [emqx_types:route()]).
+-spec(lookup_routes(emqx_types:topic()) -> [emqx_types:route()]).
 lookup_routes(Topic) ->
     ets:lookup(?ROUTE_TAB, Topic).
 
--spec(has_routes(emqx_topic:topic()) -> boolean()).
+-spec(has_routes(emqx_types:topic()) -> boolean()).
 has_routes(Topic) when is_binary(Topic) ->
     ets:member(?ROUTE_TAB, Topic).
 
--spec(delete_route(emqx_topic:topic()) -> ok | {error, term()}).
+-spec(delete_route(emqx_types:topic()) -> ok | {error, term()}).
 delete_route(Topic) when is_binary(Topic) ->
     delete_route(Topic, node()).
 
--spec(delete_route(emqx_topic:topic(), dest()) -> ok | {error, term()}).
+-spec(delete_route(emqx_types:topic(), dest()) -> ok | {error, term()}).
 delete_route(Topic, Dest) when is_binary(Topic) ->
     call(pick(Topic), {delete_route, Topic, Dest}).
 
--spec(do_delete_route(emqx_topic:topic()) -> ok | {error, term()}).
+-spec(do_delete_route(emqx_types:topic()) -> ok | {error, term()}).
 do_delete_route(Topic) when is_binary(Topic) ->
     do_delete_route(Topic, node()).
 
--spec(do_delete_route(emqx_topic:topic(), dest()) -> ok | {error, term()}).
+-spec(do_delete_route(emqx_types:topic(), dest()) -> ok | {error, term()}).
 do_delete_route(Topic, Dest) ->
     Route = #route{topic = Topic, dest = Dest},
     case emqx_topic:wildcard(Topic) of
@@ -169,12 +169,12 @@ do_delete_route(Topic, Dest) ->
         false -> delete_direct_route(Route)
     end.
 
--spec(topics() -> list(emqx_topic:topic())).
+-spec(topics() -> list(emqx_types:topic())).
 topics() ->
     mnesia:dirty_all_keys(?ROUTE_TAB).
 
 %% @doc Print routes to a topic
--spec(print_routes(emqx_topic:topic()) -> ok).
+-spec(print_routes(emqx_types:topic()) -> ok).
 print_routes(Topic) ->
     lists:foreach(fun(#route{topic = To, dest = Dest}) ->
                       io:format("~s -> ~s~n", [To, Dest])

apps/emqx/src/emqx_shared_sub.erl

@@ -103,18 +103,18 @@ mnesia(copy) ->
 start_link() ->
     gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
 
--spec(subscribe(emqx_topic:group(), emqx_topic:topic(), pid()) -> ok).
+-spec(subscribe(emqx_types:group(), emqx_types:topic(), pid()) -> ok).
 subscribe(Group, Topic, SubPid) when is_pid(SubPid) ->
     gen_server:call(?SERVER, {subscribe, Group, Topic, SubPid}).
 
--spec(unsubscribe(emqx_topic:group(), emqx_topic:topic(), pid()) -> ok).
+-spec(unsubscribe(emqx_types:group(), emqx_types:topic(), pid()) -> ok).
 unsubscribe(Group, Topic, SubPid) when is_pid(SubPid) ->
     gen_server:call(?SERVER, {unsubscribe, Group, Topic, SubPid}).
 
 record(Group, Topic, SubPid) ->
     #emqx_shared_subscription{group = Group, topic = Topic, subpid = SubPid}.
 
--spec(dispatch(emqx_topic:group(), emqx_topic:topic(), emqx_types:delivery())
+-spec(dispatch(emqx_types:group(), emqx_types:topic(), emqx_types:delivery())
       -> emqx_types:deliver_result()).
 dispatch(Group, Topic, Delivery) ->
     dispatch(Group, Topic, Delivery, _FailedSubs = []).

apps/emqx/src/emqx_trie.erl

@@ -77,7 +77,7 @@ mnesia(copy) ->
 %%--------------------------------------------------------------------
 
 %% @doc Insert a topic filter into the trie.
--spec(insert(emqx_topic:topic()) -> ok).
+-spec(insert(emqx_types:topic()) -> ok).
 insert(Topic) when is_binary(Topic) ->
     {TopicKey, PrefixKeys} = make_keys(Topic),
     case mnesia:wread({?TRIE, TopicKey}) of
@@ -86,7 +86,7 @@ insert(Topic) when is_binary(Topic) ->
     end.
 
 %% @doc Delete a topic filter from the trie.
--spec(delete(emqx_topic:topic()) -> ok).
+-spec(delete(emqx_types:topic()) -> ok).
 delete(Topic) when is_binary(Topic) ->
     {TopicKey, PrefixKeys} = make_keys(Topic),
     case [] =/= mnesia:wread({?TRIE, TopicKey}) of
@@ -95,7 +95,7 @@ delete(Topic) when is_binary(Topic) ->
     end.
 
 %% @doc Find trie nodes that matchs the topic name.
--spec(match(emqx_topic:topic()) -> list(emqx_topic:topic())).
+-spec(match(emqx_types:topic()) -> list(emqx_types:topic())).
 match(Topic) when is_binary(Topic) ->
     Words = emqx_topic:words(Topic),
     case emqx_topic:wildcard(Words) of

apps/emqx/src/emqx_types.erl

@@ -146,7 +146,7 @@
                         dn           => binary(),
                         atom()       => term()
                        }).
--type(clientid() :: binary()|atom()).
+-type(clientid() :: binary() | atom()).
 -type(username() :: maybe(binary())).
 -type(password() :: maybe(binary())).
 -type(peerhost() :: inet:ip_address()).
@@ -201,8 +201,8 @@
 -type(publish_result() :: [{node(), topic(), deliver_result()} |
                            {share, topic(), deliver_result()}]).
 -type(route() :: #route{}).
--type(sub_group() :: tuple() | binary()).
--type(route_entry() :: {topic(), node()} | {topic, sub_group()}).
+-type(group() :: emqx_topic:group()).
+-type(route_entry() :: {topic(), node()} | {topic, group()}).
 -type(plugin() :: #plugin{}).
 -type(command() :: #command{}).
 
@@ -215,4 +215,3 @@
                         max_heap_size => non_neg_integer(),
                         enable => boolean()
                        }).
-

apps/emqx/test/emqx_cm_locker_SUITE.erl

@@ -39,7 +39,7 @@ t_trans(_) ->
     ok = emqx_cm_locker:trans(<<"clientid">>, fun(_) -> ok end).
 
 t_lock_unlocak(_) ->
-    {true, _Nodes} = emqx_cm_locker:lock(<<"clientid">>),
-    {true, _Nodes} = emqx_cm_locker:lock(<<"clientid">>),
-    {true, _Nodes} = emqx_cm_locker:unlock(<<"clientid">>),
-    {true, _Nodes} = emqx_cm_locker:unlock(<<"clientid">>).
+    {true, _} = emqx_cm_locker:lock(<<"clientid">>),
+    {true, _} = emqx_cm_locker:lock(<<"clientid">>),
+    {true, _} = emqx_cm_locker:unlock(<<"clientid">>),
+    {true, _} = emqx_cm_locker:unlock(<<"clientid">>).

apps/emqx/test/emqx_request_handler.erl

@@ -20,8 +20,8 @@
 
 -include_lib("emqx/include/emqx_mqtt.hrl").
 
--type qos() :: emqx_mqtt_types:qos_name() | emqx_mqtt_types:qos().
--type topic() :: emqx_topic:topic().
+-type qos() :: emqx_types:qos_name() | emqx_types:qos().
+-type topic() :: emqx_types:topic().
 -type handler() :: fun((CorrData :: binary(), ReqPayload :: binary()) -> RspPayload :: binary()).
 
 -spec start_link(topic(), qos(), handler(), emqtt:options()) ->

apps/emqx/test/emqx_shared_sub_SUITE.erl

@@ -109,9 +109,9 @@ t_no_connection_nack(_) ->
 
     ExpProp = [{properties, #{'Session-Expiry-Interval' => timer:seconds(30)}}],
     {ok, SubConnPid1} = emqtt:start_link([{clientid, Subscriber1}] ++ ExpProp),
-    {ok, _Props} = emqtt:connect(SubConnPid1),
+    {ok, _Props1} = emqtt:connect(SubConnPid1),
     {ok, SubConnPid2} = emqtt:start_link([{clientid, Subscriber2}] ++ ExpProp),
-    {ok, _Props} = emqtt:connect(SubConnPid2),
+    {ok, _Props2} = emqtt:connect(SubConnPid2),
     emqtt:subscribe(SubConnPid1, ShareTopic, QoS),
     emqtt:subscribe(SubConnPid1, ShareTopic, QoS),
 

apps/emqx_authz/include/emqx_authz.hrl

@@ -14,7 +14,7 @@
 
 -type(permission() :: allow | deny).
 
--type(rule() :: {permission(), who(), action(), list(emqx_topic:topic())}).
+-type(rule() :: {permission(), who(), action(), list(emqx_types:topic())}).
 -type(rules() :: [rule()]).
 
 -type(sources() :: [map()]).

apps/emqx_authz/src/emqx_authz.erl

@@ -326,7 +326,7 @@ init_source(#{enable := false} = Source) ->Source.
 %%--------------------------------------------------------------------
 
 %% @doc Check AuthZ
--spec(authorize(emqx_types:clientinfo(), emqx_types:all(), emqx_topic:topic(), allow | deny, sources())
+-spec(authorize(emqx_types:clientinfo(), emqx_types:all(), emqx_types:topic(), allow | deny, sources())
       -> {stop, allow} | {ok, deny}).
 authorize(#{username := Username,
             peerhost := IpAddress

apps/emqx_connector/src/mqtt/emqx_connector_mqtt_worker.erl

@@ -104,11 +104,11 @@
              ]).
 
 -type id() :: atom() | string() | pid().
--type qos() :: emqx_mqtt_types:qos().
+-type qos() :: emqx_types:qos().
 -type config() :: map().
 -type batch() :: [emqx_connector_mqtt_msg:exp_msg()].
 -type ack_ref() :: term().
--type topic() :: emqx_topic:topic().
+-type topic() :: emqx_types:topic().
 
 -include_lib("emqx/include/logger.hrl").
 -include_lib("emqx/include/emqx_mqtt.hrl").
@@ -176,7 +176,7 @@ ping(Name) ->
 get_forwards(Name) -> gen_statem:call(name(Name), get_forwards, timer:seconds(1000)).
 
 %% @doc Return all subscriptions (subscription over mqtt connection to remote broker).
--spec get_subscriptions(id()) -> [{emqx_topic:topic(), qos()}].
+-spec get_subscriptions(id()) -> [{emqx_types:topic(), qos()}].
 get_subscriptions(Name) -> gen_statem:call(name(Name), get_subscriptions).
 
 callback_mode() -> [state_functions].
@@ -532,4 +532,4 @@ str(A) when is_atom(A) ->
 str(B) when is_binary(B) ->
     binary_to_list(B);
 str(S) when is_list(S) ->
-    S.
+    S.

apps/emqx_rule_engine/src/emqx_rule_funcs.erl

@@ -265,7 +265,7 @@ payload(Path) ->
 
 %% @doc Check if a topic_filter contains a specific topic
 %% TopicFilters = [{<<"t/a">>, #{qos => 0}].
--spec(contains_topic(emqx_mqtt_types:topic_filters(), emqx_types:topic())
+-spec(contains_topic(emqx_types:topic_filters(), emqx_types:topic())
         -> true | false).
 contains_topic(TopicFilters, Topic) ->
     case find_topic_filter(Topic, TopicFilters, fun eq/2) of
@@ -278,7 +278,7 @@ contains_topic(TopicFilters, Topic, QoS) ->
         _ -> false
     end.
 
--spec(contains_topic_match(emqx_mqtt_types:topic_filters(), emqx_types:topic())
+-spec(contains_topic_match(emqx_types:topic_filters(), emqx_types:topic())
         -> true | false).
 contains_topic_match(TopicFilters, Topic) ->
     case find_topic_filter(Topic, TopicFilters, fun emqx_topic:match/2) of

rebar.config

@@ -62,7 +62,7 @@
     , {getopt, "1.0.2"}
     , {snabbkaffe, {git, "https://github.com/kafka4beam/snabbkaffe.git", {tag, "0.14.1"}}}
     , {hocon, {git, "https://github.com/emqx/hocon.git", {tag, "0.17.1"}}}
-    , {emqx_http_lib, {git, "https://github.com/emqx/emqx_http_lib.git", {tag, "0.4.0"}}}
+    , {emqx_http_lib, {git, "https://github.com/emqx/emqx_http_lib.git", {tag, "0.4.1"}}}
     , {esasl, {git, "https://github.com/emqx/esasl", {tag, "0.2.0"}}}
     , {jose, {git, "https://github.com/potatosalad/erlang-jose", {tag, "1.11.1"}}}
     ]}.

rebar.config.erl

@@ -110,7 +110,7 @@ project_app_dirs() ->
 
 plugins(HasElixir) ->
     [ {relup_helper,{git,"https://github.com/emqx/relup_helper", {tag, "2.0.0"}}}
-    , {er_coap_client, {git, "https://github.com/emqx/er_coap_client", {tag, "v1.0.3"}}}
+    , {er_coap_client, {git, "https://github.com/emqx/er_coap_client", {tag, "v1.0.4"}}}
       %% emqx main project does not require port-compiler
       %% pin at root level for deterministic
     , {pc, {git, "https://github.com/emqx/port_compiler.git", {tag, "v1.11.1"}}}