Merge pull request #245 from emqtt/dev

Dev

README.md

@@ -37,16 +37,31 @@ emqttd is aimed to provide a solid, enterprise grade, extensible open-source MQT
 * Passed eclipse paho interoperability tests
 
 
-## Plugins
+## Modules
 
 * [emqttd_auth_clientid](https://github.com/emqtt/emqttd/wiki/Authentication) - Authentication with ClientIds
-* emqttd_auth_mysql - Authentication with MySQL
+* [emqttd_auth_username](https://github.com/emqtt/emqttd/wiki/Authentication) - Authentication with Username and Password
-* emqttd_auth_ldap - Authentication with LDAP
+* [emqttd_auth_ldap](https://github.com/emqtt/emqttd/wiki/Authentication) - Authentication with LDAP
-* emqttd_mod_autosub - Subscribe some topics automatically when client connected
 * [emqttd_mod_presence](https://github.com/emqtt/emqttd/wiki/Presence) - Publish presence message to $SYS topics when client connected or disconnected
+* emqttd_mod_autosub - Subscribe topics when client connected
 * [emqttd_mod_rewrite](https://github.com/emqtt/emqttd/wiki/Rewrite) - Topics rewrite like HTTP rewrite module
 
 
+## Plugins
+
+* [emqttd_plugin_template](https://github.com/emqtt/emqttd_plugin_template) - Plugin template and demo
+* [emqttd_dashboard](https://github.com/emqtt/emqttd_dashboard) - Web Dashboard
+* [emqttd_plugin_mysql](https://github.com/emqtt/emqttd_plugin_mysql) - Authentication with MySQL
+* [emqttd_plugin_pgsql](https://github.com/emqtt/emqttd_plugin_pgsql) - Authentication with PostgreSQL
+
+
+## Dashboard
+
+The broker released a simple web dashboard in 0.10.0 version.
+
+Address: http://host:18083
+
+
 ## Design
 
 ![emqttd architecture](http://emqtt.io/static/img/Architecture.png)

plugins/emqttd_plugin_template

@@ -1 +1 @@
-Subproject commit d10ee3dcdaf4e7d17f50ed0745c39628a96678e2
+Subproject commit 3d5d2ccabdde2d0381bcd17c803be5e42b3fec90

rel/files/acl.config

@@ -18,6 +18,8 @@
 %%
 %%%-----------------------------------------------------------------------------
 
+{allow, {user, "dashboard"}, subscribe, ["$SYS/#"]}.
+
 {allow, {ipaddr, "127.0.0.1"}, pubsub, ["$SYS/#", "#"]}.
 
 {deny, all, subscribe, ["$SYS/#", {eq, "#"}]}.

src/emqttd_log.erl

@@ -0,0 +1,31 @@
+%%%-----------------------------------------------------------------------------
+%%% 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 log trace.
+%%%
+%%% @end
+%%%-----------------------------------------------------------------------------
+
+%% TODO: issue #103
+
+-module(emqttd_log).
+

src/emqttd_mqueue.erl

@@ -59,7 +59,8 @@
 -export([new/3, name/1,
          is_empty/1, is_full/1,
          len/1, max_len/1,
-         in/2, out/1]).
+         in/2, out/1,
+         stats/1]).
 
 -define(LOW_WM, 0.2).
 
@@ -72,6 +73,7 @@
                  high_wm    = ?HIGH_WM,
                  max_len    = ?MAX_LEN,
                  qos0       = false,
+                 dropped    = 0,
                  alarm_fun}).
 
 -type mqueue() :: #mqueue{}.
@@ -111,6 +113,9 @@ len(#mqueue{len = Len}) -> Len.
 
 max_len(#mqueue{max_len= MaxLen}) -> MaxLen.
 
+stats(#mqueue{max_len = MaxLen, len = Len, dropped = Dropped}) ->
+    [{max_len, MaxLen}, {len, Len}, {dropped, Dropped}].
+
 %%------------------------------------------------------------------------------
 %% @doc Queue one message.
 %% @end
@@ -122,11 +127,11 @@ in(#mqtt_message{qos = ?QOS_0}, MQ = #mqueue{qos0 = false}) ->
     MQ;
 
 %% simply drop the oldest one if queue is full, improve later
-in(Msg, MQ = #mqueue{name = Name, q = Q, len = Len, max_len = MaxLen})
+in(Msg, MQ = #mqueue{q = Q, len = Len, max_len = MaxLen, dropped = Dropped})
     when Len =:= MaxLen ->
-    {{value, OldMsg}, Q2} = queue:out(Q),
+    {{value, _OldMsg}, Q2} = queue:out(Q),
-    lager:error("MQueue(~s) drop ~s", [Name, emqttd_message:format(OldMsg)]),
+    %lager:error("MQueue(~s) drop ~s", [Name, emqttd_message:format(OldMsg)]),
-    MQ#mqueue{q = queue:in(Msg, Q2)};
+    MQ#mqueue{q = queue:in(Msg, Q2), dropped = Dropped +1};
 
 in(Msg, MQ = #mqueue{q = Q, len = Len}) ->
     maybe_set_alarm(MQ#mqueue{q = queue:in(Msg, Q), len = Len + 1}).

src/emqttd_session.erl

@@ -647,21 +647,23 @@ start_collector(Session = #session{collect_interval = Interval}) ->
     TRef = erlang:send_after(Interval * 1000, self(), collect_info),
     Session#session{collect_timer = TRef}.
 
-info(#session{clean_sess        = CleanSess,
+info(#session{clean_sess      = CleanSess,
-              subscriptions     = Subscriptions,
+              subscriptions   = Subscriptions,
-              inflight_queue    = InflightQueue,
+              inflight_queue  = InflightQueue,
-              max_inflight      = MaxInflight,
+              max_inflight    = MaxInflight,
-              message_queue     = MessageQueue,
+              message_queue   = MessageQueue,
-              awaiting_rel      = AwaitingRel,
+              awaiting_rel    = AwaitingRel,
-              awaiting_ack      = AwaitingAck,
+              awaiting_ack    = AwaitingAck,
-              awaiting_comp     = AwaitingComp,
+              awaiting_comp   = AwaitingComp,
-              timestamp         = CreatedAt}) ->
+              timestamp       = CreatedAt}) ->
+    Stats = emqttd_mqueue:stats(MessageQueue),
     [{pid, self()},
      {clean_sess, CleanSess},
      {subscriptions, Subscriptions},
      {max_inflight, MaxInflight},
      {inflight_queue, length(InflightQueue)},
-     {message_queue, emqttd_mqueue:len(MessageQueue)},
+     {message_queue, proplists:get_value(len, Stats)},
+     {message_dropped, proplists:get_value(dropped, Stats)},
      {awaiting_rel, maps:size(AwaitingRel)},
      {awaiting_ack, maps:size(AwaitingAck)},
      {awaiting_comp, maps:size(AwaitingComp)},