From 750fb939b1246c4eb0d1616f0ed2474c0e2fa293 Mon Sep 17 00:00:00 2001 From: Feng Lee Date: Wed, 27 May 2015 15:32:04 +0800 Subject: [PATCH] mod presence --- apps/emqttd/include/emqttd_systop.hrl | 12 +-- apps/emqttd/src/emqttd_app.erl | 3 +- apps/emqttd/src/emqttd_broker.erl | 16 ++++ apps/emqttd/src/emqttd_client.erl | 6 +- apps/emqttd/src/emqttd_cluster.erl | 1 + apps/emqttd/src/emqttd_event.erl | 122 ------------------------ apps/emqttd/src/emqttd_metrics.erl | 8 +- apps/emqttd/src/emqttd_mod_presence.erl | 53 ++++++++++ apps/emqttd/src/emqttd_protocol.erl | 4 +- apps/emqttd/src/emqttd_stats.erl | 7 +- 10 files changed, 91 insertions(+), 141 deletions(-) delete mode 100644 apps/emqttd/src/emqttd_event.erl create mode 100644 apps/emqttd/src/emqttd_mod_presence.erl diff --git a/apps/emqttd/include/emqttd_systop.hrl b/apps/emqttd/include/emqttd_systop.hrl index d2c929273..9308ba682 100644 --- a/apps/emqttd/include/emqttd_systop.hrl +++ b/apps/emqttd/include/emqttd_systop.hrl @@ -43,8 +43,6 @@ -define(SYSTOP_CLIENTS, [ 'clients/count', % clients connected current 'clients/max' % max clients connected - %'clients/connected', - %'clients/disconnected', ]). %%------------------------------------------------------------------------------ @@ -59,12 +57,12 @@ %% $SYS Topics for Subscribers %%------------------------------------------------------------------------------ -define(SYSTOP_PUBSUB, [ - 'queues/count', % ... - 'queues/max', % ... - 'topics/count', % ... + 'topics/count', % ... 'topics/max', % ... - 'subscribers/count', % ... - 'subscribers/max' % ... + 'subscribers/count', % ... + 'subscribers/max', % ... + 'queues/count', % ... + 'queues/max' % ... ]). %%------------------------------------------------------------------------------ diff --git a/apps/emqttd/src/emqttd_app.erl b/apps/emqttd/src/emqttd_app.erl index 5a8daf60f..138808b08 100644 --- a/apps/emqttd/src/emqttd_app.erl +++ b/apps/emqttd/src/emqttd_app.erl @@ -68,8 +68,7 @@ print_vsn() -> ?PRINT("~s ~s is running now~n", [Desc, Vsn]). start_servers(Sup) -> - Servers = [{"emqttd event", emqttd_event}, - {"emqttd trace", emqttd_trace}, + Servers = [{"emqttd trace", emqttd_trace}, {"emqttd pooler", {supervisor, emqttd_pooler_sup}}, {"emqttd client manager", {supervisor, emqttd_cm_sup}}, {"emqttd session manager", {supervisor, emqttd_sm_sup}}, diff --git a/apps/emqttd/src/emqttd_broker.erl b/apps/emqttd/src/emqttd_broker.erl index bff331d9c..526d824b4 100644 --- a/apps/emqttd/src/emqttd_broker.erl +++ b/apps/emqttd/src/emqttd_broker.erl @@ -39,6 +39,9 @@ %% API Function Exports -export([start_link/0]). +%% Running nodes +-export([running_nodes/0]). + %% Event API -export([subscribe/1, notify/2]). @@ -71,6 +74,13 @@ start_link() -> gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). +%%------------------------------------------------------------------------------ +%% @doc Get running nodes +%% @end +%%------------------------------------------------------------------------------ +running_nodes() -> + mnesia:system_info(running_db_nodes). + %%------------------------------------------------------------------------------ %% @doc Subscribe broker event %% @end @@ -205,6 +215,7 @@ init([]) -> random:seed(now()), ets:new(?BROKER_TAB, [set, public, named_table]), % Create $SYS Topics + emqttd_pubsub:create(<<"$SYS/brokers">>), [ok = create_topic(Topic) || Topic <- ?SYSTOP_BROKERS], % Tick {ok, #state{started_at = os:timestamp(), tick_tref = start_tick(tick)}, hibernate}. @@ -244,6 +255,7 @@ handle_cast(_Msg, State) -> {noreply, State}. handle_info(tick, State) -> + retain(brokers), retain(version, list_to_binary(version())), retain(sysdescr, list_to_binary(sysdescr())), publish(uptime, list_to_binary(uptime(State))), @@ -266,6 +278,10 @@ code_change(_OldVsn, State, _Extra) -> create_topic(Topic) -> emqttd_pubsub:create(emqtt_topic:systop(Topic)). +retain(brokers) -> + Payload = list_to_binary(string:join([atom_to_list(N) || N <- running_nodes()], ",")), + publish(#mqtt_message{retain = true, topic = <<"$SYS/brokers">>, payload = Payload}). + retain(Topic, Payload) when is_binary(Payload) -> publish(#mqtt_message{retain = true, topic = emqtt_topic:systop(Topic), diff --git a/apps/emqttd/src/emqttd_client.erl b/apps/emqttd/src/emqttd_client.erl index 0c9e2413a..0fd1db69a 100644 --- a/apps/emqttd/src/emqttd_client.erl +++ b/apps/emqttd/src/emqttd_client.erl @@ -257,8 +257,10 @@ inc(_) -> notify(disconnected, _Reason, undefined) -> ingore; notify(disconnected, {shutdown, Reason}, ProtoState) -> - emqttd_event:notify({disconnected, emqttd_protocol:clientid(ProtoState), Reason}); + %emqttd_event:notify({disconnected, emqttd_protocol:clientid(ProtoState), Reason}); + ok; notify(disconnected, Reason, ProtoState) -> - emqttd_event:notify({disconnected, emqttd_protocol:clientid(ProtoState), Reason}). + %emqttd_event:notify({disconnected, emqttd_protocol:clientid(ProtoState), Reason}). + ok. diff --git a/apps/emqttd/src/emqttd_cluster.erl b/apps/emqttd/src/emqttd_cluster.erl index 242661119..bd5a1a2ac 100644 --- a/apps/emqttd/src/emqttd_cluster.erl +++ b/apps/emqttd/src/emqttd_cluster.erl @@ -34,6 +34,7 @@ %% @doc Get running nodes %% @end %%------------------------------------------------------------------------------ +%%TODO: remove... running_nodes() -> mnesia:system_info(running_db_nodes). diff --git a/apps/emqttd/src/emqttd_event.erl b/apps/emqttd/src/emqttd_event.erl deleted file mode 100644 index e226de5de..000000000 --- a/apps/emqttd/src/emqttd_event.erl +++ /dev/null @@ -1,122 +0,0 @@ -%%%----------------------------------------------------------------------------- -%%% Copyright (c) 2012-2015 eMQTT.IO, All Rights Reserved. -%%% -%%% Permission is hereby granted, free of charge, to any person obtaining a copy -%%% of this software and associated documentation files (the "Software"), to deal -%%% in the Software without restriction, including without limitation the rights -%%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell -%%% copies of the Software, and to permit persons to whom the Software is -%%% furnished to do so, subject to the following conditions: -%%% -%%% The above copyright notice and this permission notice shall be included in all -%%% copies or substantial portions of the Software. -%%% -%%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -%%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -%%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -%%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -%%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -%%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE -%%% SOFTWARE. -%%%----------------------------------------------------------------------------- -%%% @doc -%%% emqttd event manager. -%%% -%%% @end -%%%----------------------------------------------------------------------------- --module(emqttd_event). - --author("Feng Lee "). - --include_lib("emqtt/include/emqtt.hrl"). - -%% API Function Exports --export([start_link/0, add_handler/2, notify/1]). - -%% gen_event Function Exports --export([init/1, handle_event/2, handle_call/2, handle_info/2, - terminate/2, code_change/3]). - --record(state, {systop}). - -%%------------------------------------------------------------------------------ -%% @doc Start event manager -%% @end -%%------------------------------------------------------------------------------ --spec start_link() -> {ok, pid()} | {error, any()}. -start_link() -> - case gen_event:start_link({local, ?MODULE}) of - {ok, Pid} -> - add_handler(?MODULE, []), - {ok, Pid}; - {error, Reason} -> - {error, Reason} - end. - -add_handler(Handler, Args) -> - gen_event:add_handler(?MODULE, Handler, Args). - -notify(Event) -> - gen_event:notify(?MODULE, Event). - -%%%============================================================================= -%%% gen_event callbacks -%%%============================================================================= - -init([]) -> - SysTop = list_to_binary(lists:concat(["$SYS/brokers/", node(), "/"])), - {ok, #state{systop = SysTop}}. - -handle_event({connected, ClientId, Params}, State = #state{systop = SysTop}) -> - Topic = <>, - Msg = #mqtt_message{topic = Topic, payload = payload(connected, Params)}, - emqttd_pubsub:publish(event, Msg), - {ok, State}; - -%%TODO: Protect from undefined clientId... -handle_event({disconnected, undefined, Reason}, State = #state{systop = SysTop}) -> - {ok, State}; - -handle_event({disconnected, ClientId, Reason}, State = #state{systop = SysTop}) -> - Topic = <>, - Msg = #mqtt_message{topic = Topic, payload = payload(disconnected, Reason)}, - emqttd_pubsub:publish(event, Msg), - {ok, State}; - -handle_event({subscribed, ClientId, TopicTable}, State) -> - lager:error("TODO: subscribed ~s, ~p", [ClientId, TopicTable]), - {ok, State}; - -handle_event({unsubscribed, ClientId, Topics}, State) -> - lager:error("TODO: unsubscribed ~s, ~p", [ClientId, Topics]), - {ok, State}; - -handle_event(_Event, State) -> - {ok, State}. - -handle_call(_Request, State) -> - Reply = ok, - {ok, Reply, State}. - -handle_info(_Info, State) -> - {ok, State}. - -terminate(_Reason, _State) -> - ok. - -code_change(_OldVsn, State, _Extra) -> - {ok, State}. - -%%%============================================================================= -%%% Internal functions -%%%============================================================================= - -payload(connected, Params) -> - From = proplists:get_value(from, Params), - Proto = proplists:get_value(protocol, Params), - Sess = proplists:get_value(session, Params), - iolist_to_binary(io_lib:format("from: ~s~nprotocol: ~p~nsession: ~s", [From, Proto, Sess])); - -payload(disconnected, Reason) -> - list_to_binary(io_lib:format("reason: ~p", [Reason])). - diff --git a/apps/emqttd/src/emqttd_metrics.erl b/apps/emqttd/src/emqttd_metrics.erl index e4a7fea15..550d5fcd7 100644 --- a/apps/emqttd/src/emqttd_metrics.erl +++ b/apps/emqttd/src/emqttd_metrics.erl @@ -163,7 +163,7 @@ init([]) -> % Init metrics [create_metric(Metric) || Metric <- Metrics], % $SYS Topics for metrics - [ok = create_topic(Topic) || {_, Topic} <- Metrics], + [ok = emqttd_pubsub:create(metric_topic(Topic)) || {_, Topic} <- Metrics], % Tick to publish metrics {ok, #state{tick_tref = emqttd_broker:start_tick(tick)}, hibernate}. @@ -192,7 +192,7 @@ code_change(_OldVsn, State, _Extra) -> %%%============================================================================= publish(Metric, Val) -> - emqttd_pubsub:publish(metrics, #mqtt_message{topic = emqtt_topic:systop(Metric), + emqttd_pubsub:publish(metrics, #mqtt_message{topic = metric_topic(Metric), payload = emqttd_util:integer_to_binary(Val)}). create_metric({gauge, Name}) -> @@ -202,7 +202,7 @@ create_metric({counter, Name}) -> Schedulers = lists:seq(1, erlang:system_info(schedulers)), [ets:insert(?METRIC_TAB, {{Name, I}, 0}) || I <- Schedulers]. -create_topic(Topic) -> - emqttd_pubsub:create(emqtt_topic:systop(Topic)). +metric_topic(Metric) -> + emqtt_topic:systop(list_to_binary(lists:concat(['metrics/', Metric]))). diff --git a/apps/emqttd/src/emqttd_mod_presence.erl b/apps/emqttd/src/emqttd_mod_presence.erl new file mode 100644 index 000000000..ca66177e7 --- /dev/null +++ b/apps/emqttd/src/emqttd_mod_presence.erl @@ -0,0 +1,53 @@ +%%%----------------------------------------------------------------------------- +%%% Copyright (c) 2012-2015 eMQTT.IO, All Rights Reserved. +%%% +%%% Permission is hereby granted, free of charge, to any person obtaining a copy +%%% of this software and associated documentation files (the "Software"), to deal +%%% in the Software without restriction, including without limitation the rights +%%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +%%% copies of the Software, and to permit persons to whom the Software is +%%% furnished to do so, subject to the following conditions: +%%% +%%% The above copyright notice and this permission notice shall be included in all +%%% copies or substantial portions of the Software. +%%% +%%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +%%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +%%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +%%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +%%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +%%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +%%% SOFTWARE. +%%%----------------------------------------------------------------------------- +%%% @doc +%%% emqttd presence management module. +%%% +%%% @end +%%%----------------------------------------------------------------------------- +-module(emqttd_mod_presence). + +-include_lib("emqtt/include/emqtt.hrl"). + +-export([load/1, unload/1]). + +-export([client_connected/2, client_disconnected/2]). + +load(Opts) -> + emqttd_broker:hook(client_connected, {?MODULE, client_connected}, {?MODULE, client_connected, [Opts]}), + emqttd_broker:hook(client_disconnected, {?MODULE, client_disconnected}, {?MODULE, client_disconnected, [Opts]}), + {ok, Opts}. + +client_connected({Client, ClientId}, _Opts) -> + Topic = emqtt_topic:systop(list_to_binary(["clients/", ClientId, "/connected"])), + Payload = iolist_to_binary(mochijson2:encode([{ts, emqttd_util:timestamp()}])), + emqttd_pubsub:publish(presence, #mqtt_message{topic = Topic, payload = Payload}). + +client_disconnected({ClientId, Reason}, _Opts) -> + Topic = emqtt_topic:systop(list_to_binary(["clients/", ClientId, "/disconnected"])), + Payload = iolist_to_binary(mochijson2:encode([{reason, Reason}, {ts, emqttd_util:timestamp()}])), + emqttd_pubsub:publish(presence, #mqtt_message{topic = Topic, payload = Payload}). + +unload(_Opts) -> + emqttd_broker:unhook(client_connected, {?MODULE, client_connected}), + emqttd_broker:unhook(client_disconnected, {?MODULE, client_disconnected}). + diff --git a/apps/emqttd/src/emqttd_protocol.erl b/apps/emqttd/src/emqttd_protocol.erl index 08b24674f..30013ac3e 100644 --- a/apps/emqttd/src/emqttd_protocol.erl +++ b/apps/emqttd/src/emqttd_protocol.erl @@ -422,7 +422,7 @@ notify(connected, ReturnCode, #proto_state{peername = Peername, Params = [{from, emqttd_net:format(Peername)}, {protocol, ProtoVer}, {session, Sess}, - {connack, ReturnCode}], - emqttd_event:notify({connected, ClientId, Params}). + {connack, ReturnCode}]. + %emqttd_event:notify({connected, ClientId, Params}). diff --git a/apps/emqttd/src/emqttd_stats.erl b/apps/emqttd/src/emqttd_stats.erl index d55a5942c..0b5e65522 100644 --- a/apps/emqttd/src/emqttd_stats.erl +++ b/apps/emqttd/src/emqttd_stats.erl @@ -126,7 +126,7 @@ init([]) -> Topics = ?SYSTOP_CLIENTS ++ ?SYSTOP_SESSIONS ++ ?SYSTOP_PUBSUB, [ets:insert(?STATS_TAB, {Topic, 0}) || Topic <- Topics], % Create $SYS Topics - [ok = emqttd_pubsub:create(emqtt_topic:systop(Topic)) || Topic <- Topics], + [ok = emqttd_pubsub:create(stats_topic(Topic)) || Topic <- Topics], % Tick to publish stats {ok, #state{tick_tref = emqttd_broker:start_tick(tick)}, hibernate}. @@ -154,6 +154,9 @@ code_change(_OldVsn, State, _Extra) -> %%% Internal functions %%%============================================================================= publish(Stat, Val) -> - emqttd_pubsub:publish(stats, #mqtt_message{topic = emqtt_topic:systop(Stat), + emqttd_pubsub:publish(stats, #mqtt_message{topic = stats_topic(Stat), payload = emqttd_util:integer_to_binary(Val)}). +stats_topic(Stat) -> + emqtt_topic:systop(list_to_binary(lists:concat(['stats/', Stat]))). +