From 8f00e28576a5aac45879e8ff30173598ba759a4c Mon Sep 17 00:00:00 2001 From: Feng Lee Date: Fri, 17 Nov 2017 20:51:51 +0800 Subject: [PATCH 1/6] Improve the pubsub design and fix the race-condition issue --- src/emqttd.erl | 48 +++++----- src/emqttd_pubsub.erl | 77 +++++++++-------- src/emqttd_router.erl | 197 +++++++++++++++++++++--------------------- src/emqttd_server.erl | 196 ++++++++++++++++++++++------------------- src/emqttd_trie.erl | 96 ++++++++++---------- 5 files changed, 321 insertions(+), 293 deletions(-) diff --git a/src/emqttd.erl b/src/emqttd.erl index d4cdd8437..f012fdc23 100644 --- a/src/emqttd.erl +++ b/src/emqttd.erl @@ -31,8 +31,7 @@ unsubscribe/1, unsubscribe/2]). %% PubSub Management API --export([setqos/3, topics/0, subscriptions/1, subscribers/1, - is_subscribed/2, subscriber_down/1]). +-export([setqos/3, topics/0, subscriptions/1, subscribers/1, subscribed/2]). %% Hooks API -export([hook/4, hook/3, unhook/2, run_hooks/2, run_hooks/3]). @@ -43,14 +42,13 @@ %% Shutdown and reboot -export([shutdown/0, shutdown/1, reboot/0]). --type(subscriber() :: pid() | binary()). +-type(subid() :: binary()). + +-type(subscriber() :: pid() | subid() | {subid(), pid()}). -type(suboption() :: local | {qos, non_neg_integer()} | {share, {'$queue' | binary()}}). --type(pubsub_error() :: {error, {already_subscribed, binary()} - | {subscription_not_found, binary()}}). - --export_type([subscriber/0, suboption/0, pubsub_error/0]). +-export_type([subscriber/0, suboption/0]). -define(APP, ?MODULE). @@ -59,19 +57,19 @@ %%-------------------------------------------------------------------- %% @doc Start emqttd application. --spec(start() -> ok | {error, any()}). +-spec(start() -> ok | {error, term()}). start() -> application:start(?APP). %% @doc Stop emqttd application. --spec(stop() -> ok | {error, any()}). +-spec(stop() -> ok | {error, term()}). stop() -> application:stop(?APP). %% @doc Environment --spec(env(Key:: atom()) -> {ok, any()} | undefined). +-spec(env(Key :: atom()) -> {ok, any()} | undefined). env(Key) -> application:get_env(?APP, Key). %% @doc Get environment --spec(env(Key:: atom(), Default:: any()) -> undefined | any()). +-spec(env(Key :: atom(), Default :: any()) -> undefined | any()). env(Key, Default) -> application:get_env(?APP, Key, Default). %% @doc Is running? @@ -88,15 +86,15 @@ is_running(Node) -> %%-------------------------------------------------------------------- %% @doc Subscribe --spec(subscribe(iodata()) -> ok | {error, any()}). +-spec(subscribe(iodata()) -> ok | {error, term()}). subscribe(Topic) -> - subscribe(Topic, self()). + emqttd_server:subscribe(iolist_to_binary(Topic)). --spec(subscribe(iodata(), subscriber()) -> ok | {error, any()}). +-spec(subscribe(iodata(), subscriber()) -> ok | {error, term()}). subscribe(Topic, Subscriber) -> - subscribe(Topic, Subscriber, []). + emqttd_server:subscribe(iolist_to_binary(Topic), Subscriber). --spec(subscribe(iodata(), subscriber(), [suboption()]) -> ok | pubsub_error()). +-spec(subscribe(iodata(), subscriber(), [suboption()]) -> ok | {error, term()}). subscribe(Topic, Subscriber, Options) -> emqttd_server:subscribe(iolist_to_binary(Topic), Subscriber, Options). @@ -106,11 +104,11 @@ publish(Msg) -> emqttd_server:publish(Msg). %% @doc Unsubscribe --spec(unsubscribe(iodata()) -> ok | pubsub_error()). +-spec(unsubscribe(iodata()) -> ok | {error, term()}). unsubscribe(Topic) -> - unsubscribe(Topic, self()). + emqttd_server:unsubscribe(iolist_to_binary(Topic)). --spec(unsubscribe(iodata(), subscriber()) -> ok | pubsub_error()). +-spec(unsubscribe(iodata(), subscriber()) -> ok | {error, term()}). unsubscribe(Topic, Subscriber) -> emqttd_server:unsubscribe(iolist_to_binary(Topic), Subscriber). @@ -125,17 +123,13 @@ topics() -> emqttd_router:topics(). subscribers(Topic) -> emqttd_server:subscribers(iolist_to_binary(Topic)). --spec(subscriptions(subscriber()) -> [{binary(), binary(), list(suboption())}]). +-spec(subscriptions(subscriber()) -> [{emqttd:subscriber(), binary(), list(emqttd:suboption())}]). subscriptions(Subscriber) -> emqttd_server:subscriptions(Subscriber). --spec(is_subscribed(iodata(), subscriber()) -> boolean()). -is_subscribed(Topic, Subscriber) -> - emqttd_server:is_subscribed(iolist_to_binary(Topic), Subscriber). - --spec(subscriber_down(subscriber()) -> ok). -subscriber_down(Subscriber) -> - emqttd_server:subscriber_down(Subscriber). +-spec(subscribed(iodata(), subscriber()) -> boolean()). +subscribed(Topic, Subscriber) -> + emqttd_server:subscribed(iolist_to_binary(Topic), Subscriber). %%-------------------------------------------------------------------- %% Hooks API diff --git a/src/emqttd_pubsub.erl b/src/emqttd_pubsub.erl index d976618cd..994ef6230 100644 --- a/src/emqttd_pubsub.erl +++ b/src/emqttd_pubsub.erl @@ -46,7 +46,7 @@ %% Start PubSub %%-------------------------------------------------------------------- --spec(start_link(atom(), pos_integer(), list()) -> {ok, pid()} | ignore | {error, any()}). +-spec(start_link(atom(), pos_integer(), list()) -> {ok, pid()} | ignore | {error, term()}). start_link(Pool, Id, Env) -> gen_server2:start_link({local, ?PROC_NAME(?MODULE, Id)}, ?MODULE, [Pool, Id, Env], []). @@ -54,7 +54,7 @@ start_link(Pool, Id, Env) -> %% PubSub API %%-------------------------------------------------------------------- -%% @doc Subscribe a Topic +%% @doc Subscribe to a Topic -spec(subscribe(binary(), emqttd:subscriber(), [emqttd:suboption()]) -> ok). subscribe(Topic, Subscriber, Options) -> call(pick(Topic), {subscribe, Topic, Subscriber, Options}). @@ -63,8 +63,8 @@ subscribe(Topic, Subscriber, Options) -> async_subscribe(Topic, Subscriber, Options) -> cast(pick(Topic), {subscribe, Topic, Subscriber, Options}). -%% @doc Publish MQTT Message to Topic --spec(publish(binary(), any()) -> {ok, mqtt_delivery()} | ignore). +%% @doc Publish MQTT Message to Topic. +-spec(publish(binary(), mqtt_message()) -> {ok, mqtt_delivery()} | ignore). publish(Topic, Msg) -> route(lists:append(emqttd_router:match(Topic), emqttd_router:match_local(Topic)), delivery(Msg)). @@ -72,7 +72,7 @@ publish(Topic, Msg) -> route([], #mqtt_delivery{message = #mqtt_message{topic = Topic}}) -> dropped(Topic), ignore; -%% Dispatch on the local node +%% Dispatch on the local node. route([#mqtt_route{topic = To, node = Node}], Delivery = #mqtt_delivery{flows = Flows}) when Node =:= node() -> dispatch(To, Delivery#mqtt_delivery{flows = [{route, Node, To} | Flows]}); @@ -82,8 +82,8 @@ route([#mqtt_route{topic = To, node = Node}], Delivery = #mqtt_delivery{flows = forward(Node, To, Delivery#mqtt_delivery{flows = [{route, Node, To}|Flows]}); route(Routes, Delivery) -> - {ok, lists:foldl(fun(Route, DelAcc) -> - {ok, DelAcc1} = route([Route], DelAcc), DelAcc1 + {ok, lists:foldl(fun(Route, Acc) -> + {ok, Acc1} = route([Route], Acc), Acc1 end, Delivery, Routes)}. delivery(Msg) -> #mqtt_delivery{sender = self(), message = Msg, flows = []}. @@ -92,7 +92,7 @@ delivery(Msg) -> #mqtt_delivery{sender = self(), message = Msg, flows = []}. forward(Node, To, Delivery) -> rpc:cast(Node, ?PUBSUB, dispatch, [To, Delivery]), {ok, Delivery}. -%% @doc Dispatch Message to Subscribers +%% @doc Dispatch Message to Subscribers. -spec(dispatch(binary(), mqtt_delivery()) -> mqtt_delivery()). dispatch(Topic, Delivery = #mqtt_delivery{message = Msg, flows = Flows}) -> case subscribers(Topic) of @@ -107,16 +107,16 @@ dispatch(Topic, Delivery = #mqtt_delivery{message = Msg, flows = Flows}) -> {ok, Delivery#mqtt_delivery{flows = Flows1}} end. -dispatch(Pid, Topic, Msg) when is_pid(Pid) -> - Pid ! {dispatch, Topic, Msg}; -dispatch(SubId, Topic, Msg) when is_binary(SubId) -> - emqttd_sm:dispatch(SubId, Topic, Msg); -dispatch({_Share, [Sub]}, Topic, Msg) -> +%%TODO: Is SubPid aliving??? +dispatch(SubPid, Topic, Msg) when is_pid(SubPid) -> + SubPid ! {dispatch, Topic, Msg}; +dispatch({SubId, SubPid}, Topic, Msg) when is_binary(SubId), is_pid(SubPid) -> + SubPid ! {dispatch, Topic, Msg}; +dispatch({{share, _Share}, [Sub]}, Topic, Msg) -> dispatch(Sub, Topic, Msg); -dispatch({_Share, []}, _Topic, _Msg) -> +dispatch({{share, _Share}, []}, _Topic, _Msg) -> ok; -%%TODO: round-robbin -dispatch({_Share, Subs}, Topic, Msg) -> +dispatch({{share, _Share}, Subs}, Topic, Msg) -> %% round-robbin? dispatch(lists:nth(rand:uniform(length(Subs)), Subs), Topic, Msg). subscribers(Topic) -> @@ -126,8 +126,8 @@ group_by_share([]) -> []; group_by_share(Subscribers) -> {Subs1, Shares1} = - lists:foldl(fun({Share, Sub}, {Subs, Shares}) -> - {Subs, dict:append(Share, Sub, Shares)}; + lists:foldl(fun({share, Share, Sub}, {Subs, Shares}) -> + {Subs, dict:append({share, Share}, Sub, Shares)}; (Sub, {Subs, Shares}) -> {[Sub|Subs], Shares} end, {[], dict:new()}, Subscribers), @@ -155,8 +155,8 @@ call(PubSub, Req) when is_pid(PubSub) -> cast(PubSub, Msg) when is_pid(PubSub) -> gen_server2:cast(PubSub, Msg). -pick(Subscriber) -> - gproc_pool:pick_worker(pubsub, Subscriber). +pick(Topic) -> + gproc_pool:pick_worker(pubsub, Topic). %%-------------------------------------------------------------------- %% gen_server Callbacks @@ -169,22 +169,22 @@ init([Pool, Id, Env]) -> handle_call({subscribe, Topic, Subscriber, Options}, _From, State) -> add_subscriber(Topic, Subscriber, Options), - {reply, ok, setstats(State), hibernate}; + reply(ok, setstats(State)); handle_call({unsubscribe, Topic, Subscriber, Options}, _From, State) -> del_subscriber(Topic, Subscriber, Options), - {reply, ok, setstats(State), hibernate}; + reply(ok, setstats(State)); handle_call(Req, _From, State) -> ?UNEXPECTED_REQ(Req, State). handle_cast({subscribe, Topic, Subscriber, Options}, State) -> add_subscriber(Topic, Subscriber, Options), - {noreply, setstats(State), hibernate}; + noreply(setstats(State)); handle_cast({unsubscribe, Topic, Subscriber, Options}, State) -> del_subscriber(Topic, Subscriber, Options), - {noreply, setstats(State), hibernate}; + noreply(setstats(State)); handle_cast(Msg, State) -> ?UNEXPECTED_MSG(Msg, State). @@ -205,39 +205,48 @@ code_change(_OldVsn, State, _Extra) -> add_subscriber(Topic, Subscriber, Options) -> Share = proplists:get_value(share, Options), case ?is_local(Options) of - false -> add_subscriber_(Share, Topic, Subscriber); - true -> add_local_subscriber_(Share, Topic, Subscriber) + false -> add_global_subscriber(Share, Topic, Subscriber); + true -> add_local_subscriber(Share, Topic, Subscriber) end. -add_subscriber_(Share, Topic, Subscriber) -> - (not ets:member(mqtt_subscriber, Topic)) andalso emqttd_router:add_route(Topic), +add_global_subscriber(Share, Topic, Subscriber) -> + case ets:member(mqtt_subscriber, Topic) and emqttd_router:has_route(Topic) of + true -> ok; + false -> emqttd_router:add_route(Topic) + end, ets:insert(mqtt_subscriber, {Topic, shared(Share, Subscriber)}). -add_local_subscriber_(Share, Topic, Subscriber) -> +add_local_subscriber(Share, Topic, Subscriber) -> (not ets:member(mqtt_subscriber, {local, Topic})) andalso emqttd_router:add_local_route(Topic), ets:insert(mqtt_subscriber, {{local, Topic}, shared(Share, Subscriber)}). del_subscriber(Topic, Subscriber, Options) -> Share = proplists:get_value(share, Options), case ?is_local(Options) of - false -> del_subscriber_(Share, Topic, Subscriber); - true -> del_local_subscriber_(Share, Topic, Subscriber) + false -> del_global_subscriber(Share, Topic, Subscriber); + true -> del_local_subscriber(Share, Topic, Subscriber) end. -del_subscriber_(Share, Topic, Subscriber) -> +del_global_subscriber(Share, Topic, Subscriber) -> ets:delete_object(mqtt_subscriber, {Topic, shared(Share, Subscriber)}), (not ets:member(mqtt_subscriber, Topic)) andalso emqttd_router:del_route(Topic). -del_local_subscriber_(Share, Topic, Subscriber) -> +del_local_subscriber(Share, Topic, Subscriber) -> ets:delete_object(mqtt_subscriber, {{local, Topic}, shared(Share, Subscriber)}), (not ets:member(mqtt_subscriber, {local, Topic})) andalso emqttd_router:del_local_route(Topic). shared(undefined, Subscriber) -> Subscriber; shared(Share, Subscriber) -> - {Share, Subscriber}. + {share, Share, Subscriber}. setstats(State) -> emqttd_stats:setstats('subscribers/count', 'subscribers/max', ets:info(mqtt_subscriber, size)), State. +reply(Reply, State) -> + {reply, Reply, State, hibernate}. + +noreply(State) -> + {noreply, State, hibernate}. + diff --git a/src/emqttd_router.erl b/src/emqttd_router.erl index b3dd8b4ad..4d4e22160 100644 --- a/src/emqttd_router.erl +++ b/src/emqttd_router.erl @@ -28,15 +28,17 @@ -boot_mnesia({mnesia, [boot]}). -copy_mnesia({mnesia, [copy]}). -%% Start/Stop --export([start_link/0, topics/0, local_topics/0, stop/0]). +-export([start_link/0, topics/0, local_topics/0]). + +%% For eunit tests +-export([start/0, stop/0]). %% Route APIs --export([add_route/1, add_route/2, add_routes/1, match/1, print/1, - del_route/1, del_route/2, del_routes/1, has_route/1]). +-export([add_route/1, del_route/1, match/1, print/1, has_route/1]). %% Local Route API --export([add_local_route/1, del_local_route/1, match_local/1]). +-export([get_local_routes/0, add_local_route/1, match_local/1, + del_local_route/1, clean_local_routes/0]). %% gen_server Function Exports -export([init/1, handle_call/3, handle_cast/2, handle_info/2, @@ -55,10 +57,6 @@ %%-------------------------------------------------------------------- mnesia(boot) -> - ok = ekka_mnesia:create_table(mqtt_topic, [ - {ram_copies, [node()]}, - {record_name, mqtt_topic}, - {attributes, record_info(fields, mqtt_topic)}]), ok = ekka_mnesia:create_table(mqtt_route, [ {type, bag}, {ram_copies, [node()]}, @@ -66,7 +64,6 @@ mnesia(boot) -> {attributes, record_info(fields, mqtt_route)}]); mnesia(copy) -> - ok = ekka_mnesia:copy_table(mqtt_topic), ok = ekka_mnesia:copy_table(mqtt_route, ram_copies). %%-------------------------------------------------------------------- @@ -77,19 +74,26 @@ start_link() -> gen_server:start_link({local, ?ROUTER}, ?MODULE, [], []). %%-------------------------------------------------------------------- -%% API +%% Topics %%-------------------------------------------------------------------- +-spec(topics() -> list(binary())). topics() -> mnesia:dirty_all_keys(mqtt_route). +-spec(local_topics() -> list(binary())). local_topics() -> ets:select(mqtt_local_route, [{{'$1', '_'}, [], ['$1']}]). +%%-------------------------------------------------------------------- +%% Match API +%%-------------------------------------------------------------------- + %% @doc Match Routes. -spec(match(Topic:: binary()) -> [mqtt_route()]). match(Topic) when is_binary(Topic) -> - Matched = mnesia:async_dirty(fun emqttd_trie:match/1, [Topic]), + %% Optimize: ets??? + Matched = mnesia:ets(fun emqttd_trie:match/1, [Topic]), %% Optimize: route table will be replicated to all nodes. lists:append([ets:lookup(mqtt_route, To) || To <- [Topic | Matched]]). @@ -99,93 +103,68 @@ print(Topic) -> [io:format("~s -> ~s~n", [To, Node]) || #mqtt_route{topic = To, node = Node} <- match(Topic)]. -%% @doc Add Route --spec(add_route(binary() | mqtt_route()) -> ok | {error, Reason :: any()}). +%%-------------------------------------------------------------------- +%% Route Management API +%%-------------------------------------------------------------------- + +%% @doc Add Route. +-spec(add_route(binary() | mqtt_route()) -> ok | {error, Reason :: term()}). add_route(Topic) when is_binary(Topic) -> add_route(#mqtt_route{topic = Topic, node = node()}); -add_route(Route) when is_record(Route, mqtt_route) -> - add_routes([Route]). - --spec(add_route(Topic :: binary(), Node :: node()) -> ok | {error, Reason :: any()}). -add_route(Topic, Node) when is_binary(Topic), is_atom(Node) -> - add_route(#mqtt_route{topic = Topic, node = Node}). - -%% @doc Add Routes --spec(add_routes([mqtt_route()]) -> ok | {error, Reason :: any()}). -add_routes(Routes) -> - AddFun = fun() -> [add_route_(Route) || Route <- Routes] end, - case mnesia:is_transaction() of - true -> AddFun(); - false -> trans(AddFun) +add_route(Route = #mqtt_route{topic = Topic}) -> + case emqttd_topic:wildcard(Topic) of + true -> case mnesia:is_transaction() of + true -> add_trie_route(Route); + false -> trans(fun add_trie_route/1, [Route]) + end; + false -> add_direct_route(Route) end. -%% @private -add_route_(Route = #mqtt_route{topic = Topic}) -> +add_direct_route(Route) -> + mnesia:async_dirty(fun mnesia:write/1, [Route]). + +add_trie_route(Route = #mqtt_route{topic = Topic}) -> case mnesia:wread({mqtt_route, Topic}) of - [] -> - case emqttd_topic:wildcard(Topic) of - true -> emqttd_trie:insert(Topic); - false -> ok - end, - mnesia:write(Route), - mnesia:write(#mqtt_topic{topic = Topic}); - Records -> - case lists:member(Route, Records) of - true -> ok; - false -> mnesia:write(Route) - end - end. + [] -> emqttd_trie:insert(Topic); + _ -> ok + end, + mnesia:write(Route). %% @doc Delete Route --spec(del_route(binary() | mqtt_route()) -> ok | {error, Reason :: any()}). +-spec(del_route(binary() | mqtt_route()) -> ok | {error, Reason :: term()}). del_route(Topic) when is_binary(Topic) -> del_route(#mqtt_route{topic = Topic, node = node()}); -del_route(Route) when is_record(Route, mqtt_route) -> - del_routes([Route]). - --spec(del_route(Topic :: binary(), Node :: node()) -> ok | {error, Reason :: any()}). -del_route(Topic, Node) when is_binary(Topic), is_atom(Node) -> - del_route(#mqtt_route{topic = Topic, node = Node}). - -%% @doc Delete Routes --spec(del_routes([mqtt_route()]) -> ok | {error, any()}). -del_routes(Routes) -> - DelFun = fun() -> [del_route_(Route) || Route <- Routes] end, - case mnesia:is_transaction() of - true -> DelFun(); - false -> trans(DelFun) +del_route(Route = #mqtt_route{topic = Topic}) -> + case emqttd_topic:wildcard(Topic) of + true -> case mnesia:is_transaction() of + true -> del_trie_route(Route); + false -> trans(fun del_trie_route/1, [Route]) + end; + false -> del_direct_route(Route) end. -del_route_(Route = #mqtt_route{topic = Topic}) -> +del_direct_route(Route) -> + mnesia:async_dirty(fun mnesia:delete_object/1, [Route]). + +del_trie_route(Route = #mqtt_route{topic = Topic}) -> case mnesia:wread({mqtt_route, Topic}) of - [] -> - ok; - [Route] -> - %% Remove route and trie - mnesia:delete_object(Route), - case emqttd_topic:wildcard(Topic) of - true -> emqttd_trie:delete(Topic); - false -> ok - end, - mnesia:delete({mqtt_topic, Topic}); - _More -> - %% Remove route only - mnesia:delete_object(Route) + [Route] -> %% Remove route and trie + mnesia:delete_object(Route), + emqttd_trie:delete(Topic); + [_|_] -> %% Remove route only + mnesia:delete_object(Route); + [] -> ok end. -%% @doc Has Route? +%% @doc Has route? -spec(has_route(binary()) -> boolean()). -has_route(Topic) -> - Routes = case mnesia:is_transaction() of - true -> mnesia:read(mqtt_route, Topic); - false -> mnesia:dirty_read(mqtt_route, Topic) - end, - length(Routes) > 0. +has_route(Topic) when is_binary(Topic) -> + ets:member(mqtt_route, Topic). %% @private --spec(trans(function()) -> ok | {error, any()}). -trans(Fun) -> - case mnesia:transaction(Fun) of +-spec(trans(function(), list(any())) -> ok | {error, term()}). +trans(Fun, Args) -> + case mnesia:transaction(Fun, Args) of {atomic, _} -> ok; {aborted, Error} -> {error, Error} end. @@ -194,24 +173,44 @@ trans(Fun) -> %% Local Route API %%-------------------------------------------------------------------- +-spec(get_local_routes() -> list({binary(), node()})). +get_local_routes() -> + ets:tab2list(mqtt_local_route). + -spec(add_local_route(binary()) -> ok). add_local_route(Topic) -> - gen_server:cast(?ROUTER, {add_local_route, Topic}). + gen_server:call(?ROUTER, {add_local_route, Topic}). -spec(del_local_route(binary()) -> ok). del_local_route(Topic) -> - gen_server:cast(?ROUTER, {del_local_route, Topic}). + gen_server:call(?ROUTER, {del_local_route, Topic}). -spec(match_local(binary()) -> [mqtt_route()]). match_local(Name) -> - [#mqtt_route{topic = {local, Filter}, node = Node} - || {Filter, Node} <- ets:tab2list(mqtt_local_route), - emqttd_topic:match(Name, Filter)]. + case ets:info(mqtt_local_route, size) of + 0 -> []; + _ -> ets:foldl( + fun({Filter, Node}, Matched) -> + case emqttd_topic:match(Name, Filter) of + true -> [#mqtt_route{topic = {local, Filter}, node = Node} | Matched]; + false -> Matched + end + end, [], mqtt_local_route) + end. + +-spec(clean_local_routes() -> ok). +clean_local_routes() -> + gen_server:call(?ROUTER, clean_local_routes). dump() -> [{route, ets:tab2list(mqtt_route)}, {local_route, ets:tab2list(mqtt_local_route)}]. -stop() -> gen_server:call(?ROUTER, stop). +%% For unit test. +start() -> + gen_server:start({local, ?ROUTER}, ?MODULE, [], []). + +stop() -> + gen_server:call(?ROUTER, stop). %%-------------------------------------------------------------------- %% gen_server Callbacks @@ -223,21 +222,25 @@ init([]) -> {ok, TRef} = timer:send_interval(timer:seconds(1), stats), {ok, #state{stats_timer = TRef}}. +handle_call({add_local_route, Topic}, _From, State) -> + %% why node()...? + ets:insert(mqtt_local_route, {Topic, node()}), + {reply, ok, State}; + +handle_call({del_local_route, Topic}, _From, State) -> + ets:delete(mqtt_local_route, Topic), + {reply, ok, State}; + +handle_call(clean_local_routes, _From, State) -> + ets:delete_all_objects(mqtt_local_route), + {reply, ok, State}; + handle_call(stop, _From, State) -> {stop, normal, ok, State}; handle_call(_Req, _From, State) -> {reply, ignore, State}. -handle_cast({add_local_route, Topic}, State) -> - %% why node()...? - ets:insert(mqtt_local_route, {Topic, node()}), - {noreply, State}; - -handle_cast({del_local_route, Topic}, State) -> - ets:delete(mqtt_local_route, Topic), - {noreply, State}; - handle_cast(_Msg, State) -> {noreply, State}. diff --git a/src/emqttd_server.erl b/src/emqttd_server.erl index 69d18e1e4..4e05c00aa 100644 --- a/src/emqttd_server.erl +++ b/src/emqttd_server.erl @@ -37,8 +37,7 @@ async_unsubscribe/1, async_unsubscribe/2]). %% Management API. --export([setqos/3, subscriptions/1, subscribers/1, is_subscribed/2, - subscriber_down/1]). +-export([setqos/3, subscriptions/1, subscribers/1, subscribed/2]). %% Debug API -export([dump/0]). @@ -47,10 +46,10 @@ -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). --record(state, {pool, id, env, submon :: emqttd_pmon:pmon()}). +-record(state, {pool, id, env, subids :: map(), submon :: emqttd_pmon:pmon()}). -%% @doc Start server --spec(start_link(atom(), pos_integer(), list()) -> {ok, pid()} | ignore | {error, any()}). +%% @doc Start the server +-spec(start_link(atom(), pos_integer(), list()) -> {ok, pid()} | ignore | {error, term()}). start_link(Pool, Id, Env) -> gen_server2:start_link({local, ?PROC_NAME(?MODULE, Id)}, ?MODULE, [Pool, Id, Env], []). @@ -58,21 +57,21 @@ start_link(Pool, Id, Env) -> %% PubSub API %%-------------------------------------------------------------------- -%% @doc Subscribe a Topic --spec(subscribe(binary()) -> ok | emqttd:pubsub_error()). +%% @doc Subscribe to a Topic. +-spec(subscribe(binary()) -> ok | {error, term()}). subscribe(Topic) when is_binary(Topic) -> subscribe(Topic, self()). --spec(subscribe(binary(), emqttd:subscriber()) -> ok | emqttd:pubsub_error()). +-spec(subscribe(binary(), emqttd:subscriber()) -> ok | {error, term()}). subscribe(Topic, Subscriber) when is_binary(Topic) -> subscribe(Topic, Subscriber, []). -spec(subscribe(binary(), emqttd:subscriber(), [emqttd:suboption()]) -> - ok | emqttd:pubsub_error()). + ok | {error, term()}). subscribe(Topic, Subscriber, Options) when is_binary(Topic) -> - call(pick(Subscriber), {subscribe, Topic, Subscriber, Options}). + call(pick(Subscriber), {subscribe, Topic, with_subpid(Subscriber), Options}). -%% @doc Subscribe a Topic Asynchronously +%% @doc Subscribe to a Topic asynchronously. -spec(async_subscribe(binary()) -> ok). async_subscribe(Topic) when is_binary(Topic) -> async_subscribe(Topic, self()). @@ -83,7 +82,7 @@ async_subscribe(Topic, Subscriber) when is_binary(Topic) -> -spec(async_subscribe(binary(), emqttd:subscriber(), [emqttd:suboption()]) -> ok). async_subscribe(Topic, Subscriber, Options) when is_binary(Topic) -> - cast(pick(Subscriber), {subscribe, Topic, Subscriber, Options}). + cast(pick(Subscriber), {subscribe, Topic, with_subpid(Subscriber), Options}). %% @doc Publish message to Topic. -spec(publish(mqtt_message()) -> {ok, mqtt_delivery()} | ignore). @@ -109,14 +108,14 @@ trace(publish, From, #mqtt_message{topic = Topic, payload = Payload}) -> "~s PUBLISH to ~s: ~p", [From, Topic, Payload]). %% @doc Unsubscribe --spec(unsubscribe(binary()) -> ok | emqttd:pubsub_error()). +-spec(unsubscribe(binary()) -> ok | {error, term()}). unsubscribe(Topic) when is_binary(Topic) -> unsubscribe(Topic, self()). %% @doc Unsubscribe --spec(unsubscribe(binary(), emqttd:subscriber()) -> ok | emqttd:pubsub_error()). +-spec(unsubscribe(binary(), emqttd:subscriber()) -> ok | {error, term()}). unsubscribe(Topic, Subscriber) when is_binary(Topic) -> - call(pick(Subscriber), {unsubscribe, Topic, Subscriber}). + call(pick(Subscriber), {unsubscribe, Topic, with_subpid(Subscriber)}). %% @doc Async Unsubscribe -spec(async_unsubscribe(binary()) -> ok). @@ -125,32 +124,47 @@ async_unsubscribe(Topic) when is_binary(Topic) -> -spec(async_unsubscribe(binary(), emqttd:subscriber()) -> ok). async_unsubscribe(Topic, Subscriber) when is_binary(Topic) -> - cast(pick(Subscriber), {unsubscribe, Topic, Subscriber}). + cast(pick(Subscriber), {unsubscribe, Topic, with_subpid(Subscriber)}). +-spec(setqos(binary(), emqttd:subscriber(), mqtt_qos()) -> ok). setqos(Topic, Subscriber, Qos) when is_binary(Topic) -> - call(pick(Subscriber), {setqos, Topic, Subscriber, Qos}). + call(pick(Subscriber), {setqos, Topic, with_subpid(Subscriber), Qos}). --spec(subscriptions(emqttd:subscriber()) -> [{binary(), binary(), list(emqttd:suboption())}]). -subscriptions(Subscriber) -> - lists:map(fun({_, {_Share, Topic}}) -> - subscription(Topic, Subscriber); - ({_, Topic}) -> - subscription(Topic, Subscriber) - end, ets:lookup(mqtt_subscription, Subscriber)). +with_subpid(SubPid) when is_pid(SubPid) -> + SubPid; +with_subpid(SubId) when is_binary(SubId) -> + {SubId, self()}; +with_subpid({SubId, SubPid}) when is_binary(SubId), is_pid(SubPid) -> + {SubId, SubPid}. -subscription(Topic, Subscriber) -> - {Topic, Subscriber, ets:lookup_element(mqtt_subproperty, {Topic, Subscriber}, 2)}. +-spec(subscriptions(emqttd:subscriber()) -> [{emqttd:subscriber(), binary(), list(emqttd:suboption())}]). +subscriptions(SubPid) when is_pid(SubPid) -> + with_subproperty(ets:lookup(mqtt_subscription, SubPid)); -subscribers(Topic) -> +subscriptions(SubId) when is_binary(SubId) -> + with_subproperty(ets:match_object(mqtt_subscription, {{SubId, '_'}, '_'})); + +subscriptions({SubId, SubPid}) when is_binary(SubId), is_pid(SubPid) -> + with_subproperty(ets:lookup(mqtt_subscription, {SubId, SubPid})). + +with_subproperty({Subscriber, {share, _Share, Topic}}) -> + with_subproperty({Subscriber, Topic}); +with_subproperty({Subscriber, Topic}) -> + {Subscriber, Topic, ets:lookup_element(mqtt_subproperty, {Topic, Subscriber}, 2)}; +with_subproperty(Subscriptions) when is_list(Subscriptions) -> + [with_subproperty(Subscription) || Subscription <- Subscriptions]. + +-spec(subscribers(binary()) -> list(emqttd:subscriber())). +subscribers(Topic) when is_binary(Topic) -> emqttd_pubsub:subscribers(Topic). --spec(is_subscribed(binary(), emqttd:subscriber()) -> boolean()). -is_subscribed(Topic, Subscriber) when is_binary(Topic) -> - ets:member(mqtt_subproperty, {Topic, Subscriber}). - --spec(subscriber_down(emqttd:subscriber()) -> ok). -subscriber_down(Subscriber) -> - cast(pick(Subscriber), {subscriber_down, Subscriber}). +-spec(subscribed(binary(), emqttd:subscriber()) -> boolean()). +subscribed(Topic, SubPid) when is_binary(Topic), is_pid(SubPid) -> + ets:member(mqtt_subproperty, {Topic, SubPid}); +subscribed(Topic, SubId) when is_binary(Topic), is_binary(SubId) -> + length(ets:match_object(mqtt_subproperty, {{Topic, {SubId, '_'}}, '_'}, 1)) == 1; +subscribed(Topic, {SubId, SubPid}) when is_binary(Topic), is_binary(SubId), is_pid(SubPid) -> + ets:member(mqtt_subproperty, {Topic, {SubId, SubPid}}). call(Server, Req) -> gen_server2:call(Server, Req, infinity). @@ -158,8 +172,12 @@ call(Server, Req) -> cast(Server, Msg) when is_pid(Server) -> gen_server2:cast(Server, Msg). -pick(Subscriber) -> - gproc_pool:pick_worker(server, Subscriber). +pick(SubPid) when is_pid(SubPid) -> + gproc_pool:pick_worker(server, SubPid); +pick(SubId) when is_binary(SubId) -> + gproc_pool:pick_worker(server, SubId); +pick({SubId, SubPid}) when is_binary(SubId), is_pid(SubPid) -> + pick(SubId). dump() -> [{Tab, ets:tab2list(Tab)} || Tab <- [mqtt_subproperty, mqtt_subscription, mqtt_subscriber]]. @@ -170,18 +188,20 @@ dump() -> init([Pool, Id, Env]) -> ?GPROC_POOL(join, Pool, Id), - {ok, #state{pool = Pool, id = Id, env = Env, submon = emqttd_pmon:new()}}. + State = #state{pool = Pool, id = Id, env = Env, + subids = #{}, submon = emqttd_pmon:new()}, + {ok, State, hibernate, {backoff, 2000, 2000, 20000}}. handle_call({subscribe, Topic, Subscriber, Options}, _From, State) -> - case do_subscribe_(Topic, Subscriber, Options, State) of - {ok, NewState} -> {reply, ok, setstats(NewState)}; - {error, Error} -> {reply, {error, Error}, State} + case do_subscribe(Topic, Subscriber, Options, State) of + {ok, NewState} -> reply(ok, setstats(NewState)); + {error, Error} -> reply({error, Error}, State) end; handle_call({unsubscribe, Topic, Subscriber}, _From, State) -> - case do_unsubscribe_(Topic, Subscriber, State) of - {ok, NewState} -> {reply, ok, setstats(NewState), hibernate}; - {error, Error} -> {reply, {error, Error}, State} + case do_unsubscribe(Topic, Subscriber, State) of + {ok, NewState} -> reply(ok, setstats(NewState)); + {error, Error} -> reply({error, Error}, State) end; handle_call({setqos, Topic, Subscriber, Qos}, _From, State) -> @@ -190,36 +210,37 @@ handle_call({setqos, Topic, Subscriber, Qos}, _From, State) -> [{_, Opts}] -> Opts1 = lists:ukeymerge(1, [{qos, Qos}], Opts), ets:insert(mqtt_subproperty, {Key, Opts1}), - {reply, ok, State}; + reply(ok, State); [] -> - {reply, {error, {subscription_not_found, Topic}}, State} + reply({error, {subscription_not_found, Topic}}, State) end; handle_call(Req, _From, State) -> ?UNEXPECTED_REQ(Req, State). handle_cast({subscribe, Topic, Subscriber, Options}, State) -> - case do_subscribe_(Topic, Subscriber, Options, State) of - {ok, NewState} -> {noreply, setstats(NewState)}; - {error, _Error} -> {noreply, State} + case do_subscribe(Topic, Subscriber, Options, State) of + {ok, NewState} -> noreply(setstats(NewState)); + {error, _Error} -> noreply(State) end; handle_cast({unsubscribe, Topic, Subscriber}, State) -> - case do_unsubscribe_(Topic, Subscriber, State) of - {ok, NewState} -> {noreply, setstats(NewState), hibernate}; - {error, _Error} -> {noreply, State} + case do_unsubscribe(Topic, Subscriber, State) of + {ok, NewState} -> noreply(setstats(NewState)); + {error, _Error} -> noreply(State) end; -handle_cast({subscriber_down, Subscriber}, State) -> - subscriber_down_(Subscriber), - {noreply, setstats(State)}; - handle_cast(Msg, State) -> ?UNEXPECTED_MSG(Msg, State). -handle_info({'DOWN', _MRef, process, DownPid, _Reason}, State = #state{submon = PM}) -> - subscriber_down_(DownPid), - {noreply, setstats(State#state{submon = PM:erase(DownPid)}), hibernate}; +handle_info({'DOWN', _MRef, process, DownPid, _Reason}, State = #state{subids = SubIds}) -> + case maps:find(DownPid, SubIds) of + {ok, SubId} -> + clean_subscriber({SubId, DownPid}); + error -> + clean_subscriber(DownPid) + end, + noreply(setstats(demonitor_subscriber(DownPid, State))); handle_info(Info, State) -> ?UNEXPECTED_INFO(Info, State). @@ -234,62 +255,54 @@ code_change(_OldVsn, State, _Extra) -> %% Internal Functions %%-------------------------------------------------------------------- -do_subscribe_(Topic, Subscriber, Options, State) -> +do_subscribe(Topic, Subscriber, Options, State) -> case ets:lookup(mqtt_subproperty, {Topic, Subscriber}) of [] -> emqttd_pubsub:async_subscribe(Topic, Subscriber, Options), Share = proplists:get_value(share, Options), - add_subscription_(Share, Subscriber, Topic), + add_subscription(Share, Subscriber, Topic), ets:insert(mqtt_subproperty, {{Topic, Subscriber}, Options}), - {ok, monitor_subpid(Subscriber, State)}; + {ok, monitor_subscriber(Subscriber, State)}; [_] -> {error, {already_subscribed, Topic}} end. -add_subscription_(undefined, Subscriber, Topic) -> +add_subscription(undefined, Subscriber, Topic) -> ets:insert(mqtt_subscription, {Subscriber, Topic}); -add_subscription_(Share, Subscriber, Topic) -> - ets:insert(mqtt_subscription, {Subscriber, {Share, Topic}}). +add_subscription(Share, Subscriber, Topic) -> + ets:insert(mqtt_subscription, {Subscriber, {share, Share, Topic}}). -monitor_subpid(SubPid, State = #state{submon = PMon}) when is_pid(SubPid) -> - State#state{submon = PMon:monitor(SubPid)}; -monitor_subpid(_SubPid, State) -> - State. +monitor_subscriber(SubPid, State = #state{submon = SubMon}) when is_pid(SubPid) -> + State#state{submon = SubMon:monitor(SubPid)}; +monitor_subscriber({SubId, SubPid}, State = #state{subids = SubIds, submon = SubMon}) -> + State#state{subids = maps:put(SubPid, SubId, SubIds), submon = SubMon:monitor(SubPid)}. -do_unsubscribe_(Topic, Subscriber, State) -> +do_unsubscribe(Topic, Subscriber, State) -> case ets:lookup(mqtt_subproperty, {Topic, Subscriber}) of [{_, Options}] -> emqttd_pubsub:async_unsubscribe(Topic, Subscriber, Options), Share = proplists:get_value(share, Options), - del_subscription_(Share, Subscriber, Topic), + del_subscription(Share, Subscriber, Topic), ets:delete(mqtt_subproperty, {Topic, Subscriber}), - {ok, case ets:member(mqtt_subscription, Subscriber) of - true -> State; - false -> demonitor_subpid(Subscriber, State) - end}; + {ok, State}; [] -> {error, {subscription_not_found, Topic}} end. -del_subscription_(undefined, Subscriber, Topic) -> +del_subscription(undefined, Subscriber, Topic) -> ets:delete_object(mqtt_subscription, {Subscriber, Topic}); -del_subscription_(Share, Subscriber, Topic) -> - ets:delete_object(mqtt_subscription, {Subscriber, {Share, Topic}}). +del_subscription(Share, Subscriber, Topic) -> + ets:delete_object(mqtt_subscription, {Subscriber, {share, Share, Topic}}). -demonitor_subpid(SubPid, State = #state{submon = PMon}) when is_pid(SubPid) -> - State#state{submon = PMon:demonitor(SubPid)}; -demonitor_subpid(_SubPid, State) -> - State. - -subscriber_down_(Subscriber) -> - lists:foreach(fun({_, {Share, Topic}}) -> - subscriber_down_(Share, Subscriber, Topic); +clean_subscriber(Subscriber) -> + lists:foreach(fun({_, {share, Share, Topic}}) -> + clean_subscriber(Share, Subscriber, Topic); ({_, Topic}) -> - subscriber_down_(undefined, Subscriber, Topic) + clean_subscriber(undefined, Subscriber, Topic) end, ets:lookup(mqtt_subscription, Subscriber)), ets:delete(mqtt_subscription, Subscriber). -subscriber_down_(Share, Subscriber, Topic) -> +clean_subscriber(Share, Subscriber, Topic) -> case ets:lookup(mqtt_subproperty, {Topic, Subscriber}) of [] -> %% TODO:....??? @@ -300,7 +313,16 @@ subscriber_down_(Share, Subscriber, Topic) -> ets:delete(mqtt_subproperty, {Topic, Subscriber}) end. +demonitor_subscriber(SubPid, State = #state{subids = SubIds, submon = SubMon}) -> + State#state{subids = maps:remove(SubPid, SubIds), submon = SubMon:demonitor(SubPid)}. + setstats(State) -> emqttd_stats:setstats('subscriptions/count', 'subscriptions/max', ets:info(mqtt_subscription, size)), State. +reply(Reply, State) -> + {reply, Reply, State, hibernate}. + +noreply(State) -> + {noreply, State, hibernate}. + diff --git a/src/emqttd_trie.erl b/src/emqttd_trie.erl index 5b36e6e04..0bb6ec63e 100644 --- a/src/emqttd_trie.erl +++ b/src/emqttd_trie.erl @@ -31,7 +31,7 @@ -copy_mnesia({mnesia, [copy]}). %% Trie API --export([insert/1, match/1, delete/1, lookup/1]). +-export([insert/1, match/1, lookup/1, delete/1]). %%-------------------------------------------------------------------- %% Mnesia Callbacks @@ -65,22 +65,22 @@ mnesia(copy) -> -spec(insert(Topic :: binary()) -> ok). insert(Topic) when is_binary(Topic) -> case mnesia:read(mqtt_trie_node, Topic) of - [#trie_node{topic=Topic}] -> - ok; - [TrieNode=#trie_node{topic=undefined}] -> - write_trie_node(TrieNode#trie_node{topic=Topic}); - [] -> - % Add trie path - lists:foreach(fun add_path/1, emqttd_topic:triples(Topic)), - % Add last node - write_trie_node(#trie_node{node_id=Topic, topic=Topic}) + [#trie_node{topic = Topic}] -> + ok; + [TrieNode = #trie_node{topic = undefined}] -> + write_trie_node(TrieNode#trie_node{topic = Topic}); + [] -> + % Add trie path + lists:foreach(fun add_path/1, emqttd_topic:triples(Topic)), + % Add last node + write_trie_node(#trie_node{node_id = Topic, topic = Topic}) end. %% @doc Find trie nodes that match topic -spec(match(Topic :: binary()) -> list(MatchedTopic :: binary())). match(Topic) when is_binary(Topic) -> TrieNodes = match_node(root, emqttd_topic:words(Topic)), - [Name || #trie_node{topic=Name} <- TrieNodes, Name =/= undefined]. + [Name || #trie_node{topic = Name} <- TrieNodes, Name =/= undefined]. %% @doc Lookup a Trie Node -spec(lookup(NodeId :: binary()) -> [#trie_node{}]). @@ -91,13 +91,13 @@ lookup(NodeId) -> -spec(delete(Topic :: binary()) -> ok). delete(Topic) when is_binary(Topic) -> case mnesia:read(mqtt_trie_node, Topic) of - [#trie_node{edge_count=0}] -> - mnesia:delete({mqtt_trie_node, Topic}), - delete_path(lists:reverse(emqttd_topic:triples(Topic))); - [TrieNode] -> - write_trie_node(TrieNode#trie_node{topic = undefined}); - [] -> - ok + [#trie_node{edge_count = 0}] -> + mnesia:delete({mqtt_trie_node, Topic}), + delete_path(lists:reverse(emqttd_topic:triples(Topic))); + [TrieNode] -> + write_trie_node(TrieNode#trie_node{topic = undefined}); + [] -> + ok end. %%-------------------------------------------------------------------- @@ -107,19 +107,19 @@ delete(Topic) when is_binary(Topic) -> %% @private %% @doc Add path to trie tree. add_path({Node, Word, Child}) -> - Edge = #trie_edge{node_id=Node, word=Word}, + Edge = #trie_edge{node_id = Node, word = Word}, case mnesia:read(mqtt_trie_node, Node) of - [TrieNode = #trie_node{edge_count=Count}] -> - case mnesia:wread({mqtt_trie, Edge}) of - [] -> - write_trie_node(TrieNode#trie_node{edge_count=Count+1}), - write_trie(#trie{edge=Edge, node_id=Child}); - [_] -> - ok - end; - [] -> - write_trie_node(#trie_node{node_id=Node, edge_count=1}), - write_trie(#trie{edge=Edge, node_id=Child}) + [TrieNode = #trie_node{edge_count = Count}] -> + case mnesia:wread({mqtt_trie, Edge}) of + [] -> + write_trie_node(TrieNode#trie_node{edge_count = Count+1}), + write_trie(#trie{edge = Edge, node_id = Child}); + [_] -> + ok + end; + [] -> + write_trie_node(#trie_node{node_id = Node, edge_count = 1}), + write_trie(#trie{edge = Edge, node_id = Child}) end. %% @private @@ -135,20 +135,20 @@ match_node(NodeId, [], ResAcc) -> match_node(NodeId, [W|Words], ResAcc) -> lists:foldl(fun(WArg, Acc) -> - case mnesia:read(mqtt_trie, #trie_edge{node_id=NodeId, word=WArg}) of - [#trie{node_id=ChildId}] -> match_node(ChildId, Words, Acc); - [] -> Acc + case mnesia:read(mqtt_trie, #trie_edge{node_id = NodeId, word = WArg}) of + [#trie{node_id = ChildId}] -> match_node(ChildId, Words, Acc); + [] -> Acc end end, 'match_#'(NodeId, ResAcc), [W, '+']). %% @private %% @doc Match node with '#'. 'match_#'(NodeId, ResAcc) -> - case mnesia:read(mqtt_trie, #trie_edge{node_id=NodeId, word = '#'}) of - [#trie{node_id=ChildId}] -> - mnesia:read(mqtt_trie_node, ChildId) ++ ResAcc; - [] -> - ResAcc + case mnesia:read(mqtt_trie, #trie_edge{node_id = NodeId, word = '#'}) of + [#trie{node_id = ChildId}] -> + mnesia:read(mqtt_trie_node, ChildId) ++ ResAcc; + [] -> + ResAcc end. %% @private @@ -156,17 +156,17 @@ match_node(NodeId, [W|Words], ResAcc) -> delete_path([]) -> ok; delete_path([{NodeId, Word, _} | RestPath]) -> - mnesia:delete({mqtt_trie, #trie_edge{node_id=NodeId, word=Word}}), + mnesia:delete({mqtt_trie, #trie_edge{node_id = NodeId, word = Word}}), case mnesia:read(mqtt_trie_node, NodeId) of - [#trie_node{edge_count=1, topic=undefined}] -> - mnesia:delete({mqtt_trie_node, NodeId}), - delete_path(RestPath); - [TrieNode=#trie_node{edge_count=1, topic=_}] -> - write_trie_node(TrieNode#trie_node{edge_count=0}); - [TrieNode=#trie_node{edge_count=C}] -> - write_trie_node(TrieNode#trie_node{edge_count=C-1}); - [] -> - throw({notfound, NodeId}) + [#trie_node{edge_count = 1, topic = undefined}] -> + mnesia:delete({mqtt_trie_node, NodeId}), + delete_path(RestPath); + [TrieNode = #trie_node{edge_count = 1, topic = _}] -> + write_trie_node(TrieNode#trie_node{edge_count = 0}); + [TrieNode = #trie_node{edge_count = C}] -> + write_trie_node(TrieNode#trie_node{edge_count = C-1}); + [] -> + mnesia:abort({node_not_found, NodeId}) end. %% @private From dcb4a7d69baece08591abf440014c12760a2dfa8 Mon Sep 17 00:00:00 2001 From: Feng Lee Date: Fri, 17 Nov 2017 20:52:23 +0800 Subject: [PATCH 2/6] Update the test cases for emqttd_router --- test/emqttd_SUITE.erl | 25 ++++--- test/emqttd_router_SUITE.erl | 132 +++++++++++++++++++++++++++++++++++ 2 files changed, 146 insertions(+), 11 deletions(-) create mode 100644 test/emqttd_router_SUITE.erl diff --git a/test/emqttd_SUITE.erl b/test/emqttd_SUITE.erl index 195820248..e5e4f402d 100644 --- a/test/emqttd_SUITE.erl +++ b/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([{<<"$local/topic1">>,<<"x">>,[]},{<<"$local/topic2">>,<<"x">>,[{qos,2}]}], emqttd:subscriptions(<<"x">>)), + ?assertEqual([{<<"x">>, self()}], emqttd:subscribers("$local/topic1")), + ?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(), []}, - {<<"$queue/topic3">>, self(), []}, - {<<"$share/group2/topic2">>, self(), []}], + ?assertEqual([{self(), <<"$local/$share/group1/topic1">>, []}, + {self(), <<"$queue/topic3">>, []}, + {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(_) -> @@ -589,9 +592,9 @@ conflict_listeners(_) -> {current_clients, esockd:get_current_clients(Pid)}, {shutdown_count, esockd:get_shutdown_count(Pid)}]} end, esockd:listeners()), - L =proplists:get_value("mqtt:tcp:0.0.0.0:1883", 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(_) -> diff --git a/test/emqttd_router_SUITE.erl b/test/emqttd_router_SUITE.erl new file mode 100644 index 000000000..addd36288 --- /dev/null +++ b/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]). + From 62d4f9f031b384c3ad6611d31157088ca67271e7 Mon Sep 17 00:00:00 2001 From: Feng Lee Date: Fri, 17 Nov 2017 20:53:07 +0800 Subject: [PATCH 3/6] Bump version to 2.3.0 --- Makefile | 4 ++-- src/emqttd.app.src | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/Makefile b/Makefile index 3ea8fff3d..c881d41ee 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,6 @@ PROJECT = emqttd PROJECT_DESCRIPTION = Erlang MQTT Broker -PROJECT_VERSION = 2.3 +PROJECT_VERSION = 2.3.0 DEPS = goldrush gproc lager esockd ekka mochiweb pbkdf2 lager_syslog bcrypt clique jsx @@ -36,7 +36,7 @@ EUNIT_OPTS = verbose CT_SUITES = emqttd emqttd_access emqttd_lib emqttd_inflight emqttd_mod \ emqttd_net emqttd_mqueue emqttd_protocol emqttd_topic \ - emqttd_trie emqttd_vm emqttd_config + emqttd_router emqttd_trie emqttd_vm emqttd_config CT_OPTS = -cover test/ct.cover.spec -erl_args -name emqttd_ct@127.0.0.1 diff --git a/src/emqttd.app.src b/src/emqttd.app.src index 0b85fe0f0..269601bb8 100644 --- a/src/emqttd.app.src +++ b/src/emqttd.app.src @@ -1,6 +1,6 @@ {application,emqttd, [{description,"Erlang MQTT Broker"}, - {vsn,"2.3"}, + {vsn,"2.3.0"}, {modules,[]}, {registered,[emqttd_sup]}, {applications,[kernel,stdlib,gproc,lager,esockd,mochiweb, From 1545e18cd9ff0b291e9fdcd228b8a565269a3a82 Mon Sep 17 00:00:00 2001 From: Feng Lee Date: Fri, 17 Nov 2017 20:55:11 +0800 Subject: [PATCH 4/6] Remove the 'mqtt.pubsub.by_clientid' option --- etc/emq.conf | 4 +--- priv/emq.schema | 5 ----- 2 files changed, 1 insertion(+), 8 deletions(-) diff --git a/etc/emq.conf b/etc/emq.conf index 227a22457..511292d30 100644 --- a/etc/emq.conf +++ b/etc/emq.conf @@ -1,5 +1,5 @@ ##=================================================================== -## EMQ Configuration R2.3 +## EMQ Configuration R2.3.0 ##=================================================================== ##-------------------------------------------------------------------- @@ -288,8 +288,6 @@ mqtt.broker.sys_interval = 60 ## PubSub Pool Size. Default should be scheduler numbers. mqtt.pubsub.pool_size = 8 -mqtt.pubsub.by_clientid = true - ## Subscribe Asynchronously mqtt.pubsub.async = true diff --git a/priv/emq.schema b/priv/emq.schema index 6b3f0001f..1af278743 100644 --- a/priv/emq.schema +++ b/priv/emq.schema @@ -715,11 +715,6 @@ end}. {datatype, integer} ]}. -{mapping, "mqtt.pubsub.by_clientid", "emqttd.pubsub", [ - {default, true}, - {datatype, {enum, [true, false]}} -]}. - {mapping, "mqtt.pubsub.async", "emqttd.pubsub", [ {default, true}, {datatype, {enum, [true, false]}} From d41fd94abb7ca6d24b0bab323f176f63b3bb8da1 Mon Sep 17 00:00:00 2001 From: Feng Lee Date: Fri, 17 Nov 2017 20:55:57 +0800 Subject: [PATCH 5/6] Format the 'trie_node' record --- include/emqttd_trie.hrl | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/include/emqttd_trie.hrl b/include/emqttd_trie.hrl index c5b9d03c7..eb4e1390d 100644 --- a/include/emqttd_trie.hrl +++ b/include/emqttd_trie.hrl @@ -17,10 +17,10 @@ -type(trie_node_id() :: binary() | atom()). -record(trie_node, - { node_id :: trie_node_id(), - edge_count = 0 :: non_neg_integer(), - topic :: binary() | undefined, - flags :: [retained | static] + { node_id :: trie_node_id(), + edge_count = 0 :: non_neg_integer(), + topic :: binary() | undefined, + flags :: [retained | static] }). -record(trie_edge, From 11a41166d2d874fca16d8dd0fcdf4eeeb998f051 Mon Sep 17 00:00:00 2001 From: Feng Lee Date: Fri, 17 Nov 2017 21:13:20 +0800 Subject: [PATCH 6/6] Fix the function spec: '{error, any()}' -> '{error, term()}' --- src/emqttd.erl | 6 +++--- src/emqttd_access_control.erl | 10 +++++----- src/emqttd_acl_mod.erl | 2 +- src/emqttd_bridge_sup.erl | 2 +- src/emqttd_bridge_sup_sup.erl | 4 ++-- src/emqttd_broker.erl | 2 +- src/emqttd_cli.erl | 9 --------- src/emqttd_cm.erl | 2 +- src/emqttd_gen_mod.erl | 4 ++-- src/emqttd_keepalive.erl | 4 ++-- src/emqttd_metrics.erl | 4 ++-- src/emqttd_mod_sup.erl | 2 +- src/emqttd_parser.erl | 2 +- src/emqttd_plugins.erl | 8 ++++---- src/emqttd_pool_sup.erl | 4 ++-- src/emqttd_pooler.erl | 2 +- src/emqttd_protocol.erl | 2 +- src/emqttd_session.erl | 8 ++++---- src/emqttd_sm.erl | 6 +++--- src/emqttd_sm_helper.erl | 2 +- src/emqttd_trace.erl | 4 ++-- 21 files changed, 40 insertions(+), 49 deletions(-) diff --git a/src/emqttd.erl b/src/emqttd.erl index f012fdc23..65739952f 100644 --- a/src/emqttd.erl +++ b/src/emqttd.erl @@ -136,17 +136,17 @@ subscribed(Topic, Subscriber) -> %%-------------------------------------------------------------------- -spec(hook(atom(), function() | {emqttd_hooks:hooktag(), function()}, list(any())) - -> ok | {error, any()}). + -> ok | {error, term()}). hook(Hook, TagFunction, InitArgs) -> emqttd_hooks:add(Hook, TagFunction, InitArgs). -spec(hook(atom(), function() | {emqttd_hooks:hooktag(), function()}, list(any()), integer()) - -> ok | {error, any()}). + -> ok | {error, term()}). hook(Hook, TagFunction, InitArgs, Priority) -> emqttd_hooks:add(Hook, TagFunction, InitArgs, Priority). -spec(unhook(atom(), function() | {emqttd_hooks:hooktag(), function()}) - -> ok | {error, any()}). + -> ok | {error, term()}). unhook(Hook, TagFunction) -> emqttd_hooks:delete(Hook, TagFunction). diff --git a/src/emqttd_access_control.erl b/src/emqttd_access_control.erl index 6cfcc03cf..601fd263f 100644 --- a/src/emqttd_access_control.erl +++ b/src/emqttd_access_control.erl @@ -43,12 +43,12 @@ %%-------------------------------------------------------------------- %% @doc Start access control server. --spec(start_link() -> {ok, pid()} | ignore | {error, any()}). +-spec(start_link() -> {ok, pid()} | ignore | {error, term()}). start_link() -> gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). %% @doc Authenticate MQTT Client. --spec(auth(Client :: mqtt_client(), Password :: password()) -> ok | {ok, boolean()} | {error, any()}). +-spec(auth(Client :: mqtt_client(), Password :: password()) -> ok | {ok, boolean()} | {error, term()}). auth(Client, Password) when is_record(Client, mqtt_client) -> auth(Client, Password, lookup_mods(auth)). auth(_Client, _Password, []) -> @@ -88,16 +88,16 @@ reload_acl() -> [Mod:reload_acl(State) || {Mod, State, _Seq} <- lookup_mods(acl)]. %% @doc Register Authentication or ACL module. --spec(register_mod(auth | acl, atom(), list()) -> ok | {error, any()}). +-spec(register_mod(auth | acl, atom(), list()) -> ok | {error, term()}). register_mod(Type, Mod, Opts) when Type =:= auth; Type =:= acl-> register_mod(Type, Mod, Opts, 0). --spec(register_mod(auth | acl, atom(), list(), non_neg_integer()) -> ok | {error, any()}). +-spec(register_mod(auth | acl, atom(), list(), non_neg_integer()) -> ok | {error, term()}). register_mod(Type, Mod, Opts, Seq) when Type =:= auth; Type =:= acl-> gen_server:call(?SERVER, {register_mod, Type, Mod, Opts, Seq}). %% @doc Unregister authentication or ACL module --spec(unregister_mod(Type :: auth | acl, Mod :: atom()) -> ok | {error, any()}). +-spec(unregister_mod(Type :: auth | acl, Mod :: atom()) -> ok | {error, not_found | term()}). unregister_mod(Type, Mod) when Type =:= auth; Type =:= acl -> gen_server:call(?SERVER, {unregister_mod, Type, Mod}). diff --git a/src/emqttd_acl_mod.erl b/src/emqttd_acl_mod.erl index 12e949afe..4ed07b369 100644 --- a/src/emqttd_acl_mod.erl +++ b/src/emqttd_acl_mod.erl @@ -32,7 +32,7 @@ PubSub :: pubsub(), Topic :: binary()}, State :: any()) -> allow | deny | ignore). --callback(reload_acl(State :: any()) -> ok | {error, any()}). +-callback(reload_acl(State :: any()) -> ok | {error, term()}). -callback(description() -> string()). diff --git a/src/emqttd_bridge_sup.erl b/src/emqttd_bridge_sup.erl index d28b2274c..75138332f 100644 --- a/src/emqttd_bridge_sup.erl +++ b/src/emqttd_bridge_sup.erl @@ -23,7 +23,7 @@ %%-------------------------------------------------------------------- %% @doc Start bridge pool supervisor --spec(start_link(atom(), binary(), [emqttd_bridge:option()]) -> {ok, pid()} | {error, any()}). +-spec(start_link(atom(), binary(), [emqttd_bridge:option()]) -> {ok, pid()} | {error, term()}). start_link(Node, Topic, Options) -> MFA = {emqttd_bridge, start_link, [Node, Topic, Options]}, emqttd_pool_sup:start_link({bridge, Node, Topic}, random, MFA). diff --git a/src/emqttd_bridge_sup_sup.erl b/src/emqttd_bridge_sup_sup.erl index 1d47bd003..11679aba8 100644 --- a/src/emqttd_bridge_sup_sup.erl +++ b/src/emqttd_bridge_sup_sup.erl @@ -40,11 +40,11 @@ bridges() -> <- supervisor:which_children(?MODULE)]. %% @doc Start a bridge --spec(start_bridge(atom(), binary()) -> {ok, pid()} | {error, any()}). +-spec(start_bridge(atom(), binary()) -> {ok, pid()} | {error, term()}). start_bridge(Node, Topic) when is_atom(Node) andalso is_binary(Topic) -> start_bridge(Node, Topic, []). --spec(start_bridge(atom(), binary(), [emqttd_bridge:option()]) -> {ok, pid()} | {error, any()}). +-spec(start_bridge(atom(), binary(), [emqttd_bridge:option()]) -> {ok, pid()} | {error, term()}). start_bridge(Node, _Topic, _Options) when Node =:= node() -> {error, bridge_to_self}; start_bridge(Node, Topic, Options) when is_atom(Node) andalso is_binary(Topic) -> diff --git a/src/emqttd_broker.erl b/src/emqttd_broker.erl index 9f939a45e..0161720f2 100644 --- a/src/emqttd_broker.erl +++ b/src/emqttd_broker.erl @@ -61,7 +61,7 @@ %%-------------------------------------------------------------------- %% @doc Start emqttd broker --spec(start_link() -> {ok, pid()} | ignore | {error, any()}). +-spec(start_link() -> {ok, pid()} | ignore | {error, term()}). start_link() -> gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). diff --git a/src/emqttd_cli.erl b/src/emqttd_cli.erl index a4029d46f..c8836230f 100644 --- a/src/emqttd_cli.erl +++ b/src/emqttd_cli.erl @@ -248,7 +248,6 @@ subscriptions(["show", ClientId]) -> Records -> [print(subscription, Subscription) || Subscription <- Records] end; - subscriptions(["add", ClientId, Topic, QoS]) -> Add = fun(IntQos) -> case emqttd:subscribe(bin(Topic), bin(ClientId), [{qos, IntQos}]) of @@ -260,22 +259,14 @@ subscriptions(["add", ClientId, Topic, QoS]) -> end, if_valid_qos(QoS, Add); - - -subscriptions(["del", ClientId]) -> - Ok = emqttd:subscriber_down(bin(ClientId)), - ?PRINT("~p~n", [Ok]); - subscriptions(["del", ClientId, Topic]) -> Ok = emqttd:unsubscribe(bin(Topic), bin(ClientId)), ?PRINT("~p~n", [Ok]); - subscriptions(_) -> ?USAGE([{"subscriptions list", "List all subscriptions"}, {"subscriptions show ", "Show subscriptions of a client"}, {"subscriptions add ", "Add a static subscription manually"}, - {"subscriptions del ", "Delete static subscriptions manually"}, {"subscriptions del ", "Delete a static subscription manually"}]). % if_could_print(Tab, Fun) -> diff --git a/src/emqttd_cm.erl b/src/emqttd_cm.erl index 2e57ebe5b..c5f0ff145 100644 --- a/src/emqttd_cm.erl +++ b/src/emqttd_cm.erl @@ -47,7 +47,7 @@ %%-------------------------------------------------------------------- %% @doc Start Client Manager --spec(start_link(atom(), pos_integer(), fun()) -> {ok, pid()} | ignore | {error, any()}). +-spec(start_link(atom(), pos_integer(), fun()) -> {ok, pid()} | ignore | {error, term()}). start_link(Pool, Id, StatsFun) -> gen_server2:start_link(?MODULE, [Pool, Id, StatsFun], []). diff --git a/src/emqttd_gen_mod.erl b/src/emqttd_gen_mod.erl index 824b65f4c..f8d690024 100644 --- a/src/emqttd_gen_mod.erl +++ b/src/emqttd_gen_mod.erl @@ -24,9 +24,9 @@ -ifdef(use_specs). --callback(load(Opts :: any()) -> ok | {error, any()}). +-callback(load(Opts :: any()) -> ok | {error, term()}). --callback(unload(State :: any()) -> any()). +-callback(unload(State :: term()) -> term()). -else. diff --git a/src/emqttd_keepalive.erl b/src/emqttd_keepalive.erl index 5d26ea8a6..a0458038a 100644 --- a/src/emqttd_keepalive.erl +++ b/src/emqttd_keepalive.erl @@ -29,7 +29,7 @@ -export_type([keepalive/0]). %% @doc Start a keepalive --spec(start(fun(), integer(), any()) -> {ok, keepalive()} | {error, any()}). +-spec(start(fun(), integer(), any()) -> {ok, keepalive()} | {error, term()}). start(_, 0, _) -> {ok, #keepalive{}}; start(StatFun, TimeoutSec, TimeoutMsg) -> @@ -43,7 +43,7 @@ start(StatFun, TimeoutSec, TimeoutMsg) -> end. %% @doc Check keepalive, called when timeout. --spec(check(keepalive()) -> {ok, keepalive()} | {error, any()}). +-spec(check(keepalive()) -> {ok, keepalive()} | {error, term()}). check(KeepAlive = #keepalive{statfun = StatFun, statval = LastVal, repeat = Repeat}) -> case StatFun() of {ok, NewVal} -> diff --git a/src/emqttd_metrics.erl b/src/emqttd_metrics.erl index c21d274a0..17e6e96d4 100644 --- a/src/emqttd_metrics.erl +++ b/src/emqttd_metrics.erl @@ -96,8 +96,8 @@ %% API %%-------------------------------------------------------------------- -%% @doc Start metrics server --spec(start_link() -> {ok, pid()} | ignore | {error, any()}). +%% @doc Start the metrics server +-spec(start_link() -> {ok, pid()} | ignore | {error, term()}). start_link() -> gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). diff --git a/src/emqttd_mod_sup.erl b/src/emqttd_mod_sup.erl index 586d5af1a..749b84a42 100644 --- a/src/emqttd_mod_sup.erl +++ b/src/emqttd_mod_sup.erl @@ -42,7 +42,7 @@ start_child(ChildSpec) when is_tuple(ChildSpec) -> start_child(Mod, Type) when is_atom(Mod) andalso is_atom(Type) -> supervisor:start_child(?MODULE, ?CHILD(Mod, Type)). --spec(stop_child(any()) -> ok | {error, any()}). +-spec(stop_child(any()) -> ok | {error, term()}). stop_child(ChildId) -> case supervisor:terminate_child(?MODULE, ChildId) of ok -> supervisor:delete_child(?MODULE, ChildId); diff --git a/src/emqttd_parser.erl b/src/emqttd_parser.erl index 27be18a25..4699b3f77 100644 --- a/src/emqttd_parser.erl +++ b/src/emqttd_parser.erl @@ -39,7 +39,7 @@ initial_state(MaxSize) -> %% @doc Parse MQTT Packet -spec(parse(binary(), {none, pos_integer()} | fun()) - -> {ok, mqtt_packet()} | {error, any()} | {more, fun()}). + -> {ok, mqtt_packet()} | {error, term()} | {more, fun()}). parse(<<>>, {none, MaxLen}) -> {more, fun(Bin) -> parse(Bin, {none, MaxLen}) end}; parse(<>, {none, Limit}) -> diff --git a/src/emqttd_plugins.erl b/src/emqttd_plugins.erl index 73f184a3a..81ff61a4d 100644 --- a/src/emqttd_plugins.erl +++ b/src/emqttd_plugins.erl @@ -47,7 +47,7 @@ init_config(CfgFile) -> end, AppsEnv). %% @doc Load all plugins when the broker started. --spec(load() -> list() | {error, any()}). +-spec(load() -> list() | {error, term()}). load() -> case emqttd:env(plugins_loaded_file) of {ok, File} -> @@ -80,7 +80,7 @@ load_plugins(Names, Persistent) -> [load_plugin(find_plugin(Name, Plugins), Persistent) || Name <- NeedToLoad]. %% @doc Unload all plugins before broker stopped. --spec(unload() -> list() | {error, any()}). +-spec(unload() -> list() | {error, term()}). unload() -> case emqttd:env(plugins_loaded_file) of {ok, File} -> @@ -119,7 +119,7 @@ plugin(CfgFile) -> #mqtt_plugin{name = AppName, version = Ver, descr = Descr}. %% @doc Load a Plugin --spec(load(atom()) -> ok | {error, any()}). +-spec(load(atom()) -> ok | {error, term()}). load(PluginName) when is_atom(PluginName) -> case lists:member(PluginName, names(started_app)) of true -> @@ -172,7 +172,7 @@ find_plugin(Name, Plugins) -> lists:keyfind(Name, 2, Plugins). %% @doc UnLoad a Plugin --spec(unload(atom()) -> ok | {error, any()}). +-spec(unload(atom()) -> ok | {error, term()}). unload(PluginName) when is_atom(PluginName) -> case {lists:member(PluginName, names(started_app)), lists:member(PluginName, names(plugin))} of {true, true} -> diff --git a/src/emqttd_pool_sup.erl b/src/emqttd_pool_sup.erl index fa1d6aace..87654bcff 100644 --- a/src/emqttd_pool_sup.erl +++ b/src/emqttd_pool_sup.erl @@ -36,12 +36,12 @@ spec(ChildId, Args) -> {ChildId, {?MODULE, start_link, Args}, transient, infinity, supervisor, [?MODULE]}. --spec(start_link(atom() | tuple(), atom(), mfa()) -> {ok, pid()} | {error, any()}). +-spec(start_link(atom() | tuple(), atom(), mfa()) -> {ok, pid()} | {error, term()}). start_link(Pool, Type, MFA) -> Schedulers = erlang:system_info(schedulers), start_link(Pool, Type, Schedulers, MFA). --spec(start_link(atom(), atom(), pos_integer(), mfa()) -> {ok, pid()} | {error, any()}). +-spec(start_link(atom(), atom(), pos_integer(), mfa()) -> {ok, pid()} | {error, term()}). start_link(Pool, Type, Size, MFA) -> supervisor:start_link(?MODULE, [Pool, Type, Size, MFA]). diff --git a/src/emqttd_pooler.erl b/src/emqttd_pooler.erl index 1b73d138c..a74e01fec 100644 --- a/src/emqttd_pooler.erl +++ b/src/emqttd_pooler.erl @@ -40,7 +40,7 @@ start_link() -> %% API %%-------------------------------------------------------------------- --spec(start_link(atom(), pos_integer()) -> {ok, pid()} | ignore | {error, any()}). +-spec(start_link(atom(), pos_integer()) -> {ok, pid()} | ignore | {error, term()}). start_link(Pool, Id) -> gen_server:start_link({local, ?PROC_NAME(?MODULE, Id)}, ?MODULE, [Pool, Id], []). diff --git a/src/emqttd_protocol.erl b/src/emqttd_protocol.erl index 31354dd84..35c914b14 100644 --- a/src/emqttd_protocol.erl +++ b/src/emqttd_protocol.erl @@ -124,7 +124,7 @@ session(#proto_state{session = Session}) -> %% CONNECT – Client requests a connection to a Server %% A Client can only send the CONNECT Packet once over a Network Connection. --spec(received(mqtt_packet(), proto_state()) -> {ok, proto_state()} | {error, any()}). +-spec(received(mqtt_packet(), proto_state()) -> {ok, proto_state()} | {error, term()}). received(Packet = ?PACKET(?CONNECT), State = #proto_state{connected = false, stats_data = Stats}) -> trace(recv, Packet, State), Stats1 = inc_stats(recv, ?CONNECT, Stats), diff --git a/src/emqttd_session.erl b/src/emqttd_session.erl index 506743d03..aa08d746c 100644 --- a/src/emqttd_session.erl +++ b/src/emqttd_session.erl @@ -172,7 +172,7 @@ "Session(~s): " ++ Format, [State#state.client_id | Args])). %% @doc Start a Session --spec(start_link(boolean(), {mqtt_client_id(), mqtt_username()}, pid()) -> {ok, pid()} | {error, any()}). +-spec(start_link(boolean(), {mqtt_client_id(), mqtt_username()}, pid()) -> {ok, pid()} | {error, term()}). start_link(CleanSess, {ClientId, Username}, ClientPid) -> gen_server2:start_link(?MODULE, [CleanSess, {ClientId, Username}, ClientPid], []). @@ -192,7 +192,7 @@ subscribe(Session, PacketId, TopicTable) -> %%TODO: the ack function??... gen_server2:cast(Session, {subscribe, From, TopicTable, AckFun}). %% @doc Publish Message --spec(publish(pid(), mqtt_message()) -> ok | {error, any()}). +-spec(publish(pid(), mqtt_message()) -> ok | {error, term()}). publish(_Session, Msg = #mqtt_message{qos = ?QOS_0}) -> %% Publish QoS0 Directly emqttd_server:publish(Msg), ok; @@ -582,9 +582,9 @@ handle_info(Info, Session) -> ?UNEXPECTED_INFO(Info, Session). terminate(Reason, #state{client_id = ClientId, username = Username}) -> - emqttd_stats:del_session_stats(ClientId), + %% Move to emqttd_sm to avoid race condition + %%emqttd_stats:del_session_stats(ClientId), emqttd_hooks:run('session.terminated', [ClientId, Username, Reason]), - emqttd_server:subscriber_down(ClientId), emqttd_sm:unregister_session(ClientId). code_change(_OldVsn, Session, _Extra) -> diff --git a/src/emqttd_sm.erl b/src/emqttd_sm.erl index dacb0b6a1..8d4c220b2 100644 --- a/src/emqttd_sm.erl +++ b/src/emqttd_sm.erl @@ -76,12 +76,12 @@ mnesia(copy) -> %%-------------------------------------------------------------------- %% @doc Start a session manager --spec(start_link(atom(), pos_integer()) -> {ok, pid()} | ignore | {error, any()}). +-spec(start_link(atom(), pos_integer()) -> {ok, pid()} | ignore | {error, term()}). start_link(Pool, Id) -> gen_server2:start_link({local, ?PROC_NAME(?MODULE, Id)}, ?MODULE, [Pool, Id], []). %% @doc Start a session --spec(start_session(boolean(), {binary(), binary() | undefined}) -> {ok, pid(), boolean()} | {error, any()}). +-spec(start_session(boolean(), {binary(), binary() | undefined}) -> {ok, pid(), boolean()} | {error, term()}). start_session(CleanSess, {ClientId, Username}) -> SM = gproc_pool:pick_worker(?POOL, ClientId), call(SM, {start_session, CleanSess, {ClientId, Username}, self()}). @@ -107,6 +107,7 @@ unregister_session(ClientId) -> unregister_session(ClientId, Pid) -> case ets:lookup(mqtt_local_session, ClientId) of [LocalSess = {_, Pid, _, _}] -> + emqttd_stats:del_session_stats(ClientId), ets:delete_object(mqtt_local_session, LocalSess); _ -> false @@ -187,7 +188,6 @@ handle_info({'DOWN', MRef, process, DownPid, _Reason}, State) -> [] -> ok; [Sess = #mqtt_session{sess_pid = DownPid}] -> - emqttd_stats:del_session_stats(ClientId), mnesia:delete_object(mqtt_session, Sess, write); [_Sess] -> ok diff --git a/src/emqttd_sm_helper.erl b/src/emqttd_sm_helper.erl index 2e8e9a749..0721339fd 100644 --- a/src/emqttd_sm_helper.erl +++ b/src/emqttd_sm_helper.erl @@ -39,7 +39,7 @@ -define(LOCK, {?MODULE, clean_sessions}). %% @doc Start a session helper --spec(start_link(fun()) -> {ok, pid()} | ignore | {error, any()}). +-spec(start_link(fun()) -> {ok, pid()} | ignore | {error, term()}). start_link(StatsFun) -> gen_server:start_link({local, ?MODULE}, ?MODULE, [StatsFun], []). diff --git a/src/emqttd_trace.erl b/src/emqttd_trace.erl index eda84bcb1..05734c2d2 100644 --- a/src/emqttd_trace.erl +++ b/src/emqttd_trace.erl @@ -47,7 +47,7 @@ start_link() -> gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). %% @doc Start to trace client or topic. --spec(start_trace(trace_who(), string()) -> ok | {error, any()}). +-spec(start_trace(trace_who(), string()) -> ok | {error, term()}). start_trace({client, ClientId}, LogFile) -> start_trace({start_trace, {client, ClientId}, LogFile}); @@ -57,7 +57,7 @@ start_trace({topic, Topic}, LogFile) -> start_trace(Req) -> gen_server:call(?MODULE, Req, infinity). %% @doc Stop tracing client or topic. --spec(stop_trace(trace_who()) -> ok | {error, any()}). +-spec(stop_trace(trace_who()) -> ok | {error, term()}). stop_trace({client, ClientId}) -> gen_server:call(?MODULE, {stop_trace, {client, ClientId}}); stop_trace({topic, Topic}) ->