扩展插件(Plugins)

    emqtt项目组开发维护的插件包括:

    emqttd插件实际是一个Erlang应用，带自身的配置文件’etc/plugin.config”，置于’emqttd/plugins’目录下。

    plugins/emqttd_plugin_template是一个模版插件，典型目录结构:

    管理命令行’./bin/emqttd_ctl’加载卸载插件。

    加载插件:

    卸载插件:

    查询插件:

    1. ./bin/emqttd_ctl plugins list

    emqttd_dashboard: Dashboard插件

    emqttd消息服务器的Web管理控制台。插件项目地址:

    emqttd消息服务器默认加载Dashboard插件。URL地址: http://localhost:18083 ，缺省用户名/密码: admin/public。

    Dashboard插件可查询emqttd基本信息、统计数据、度量数据，查询系统客户端(Client)、会话(Session)、主题(Topic)、订阅(Subscription)。

    Dashboard插件设置

    plugins/emqttd_dashboard/etc/plugin.config:

    1. [
    2.   {emqttd_dashboard, [
    3.     {listener,
    4.       {emqttd_dashboard, 18083, [
    5.         {acceptors, 4},
    6.         {max_clients, 512}]}
    7.     }
    8.   ]}
    9. ].

    emqttd_auth_http: HTTP认证/访问控制插件

    HTTP认证/访问控制插件:

    Note

    1.1版本支持

    配置插件

    plugins/emqttd_auth_http/etc/plugin.config:

    1. [
    3.   {emqttd_auth_http, [
    5.     %% Variables: %u = username, %c = clientid, %a = ipaddress, %t = topic
    7.     {super_req, [
    8.       {method, post},
    9.       {url, "http://localhost:8080/mqtt/superuser"},
    10.       {params, [
    11.         {username, "%u"},
    12.         {clientid, "%c"}
    13.       ]}
    14.     ]},
    16.     {auth_req, [
    17.       {method, post},
    18.       {url, "http://localhost:8080/mqtt/auth"},
    19.       {params, [
    20.         {clientid, "%c"},
    21.         {username, "%u"},
    22.         {password, "%P"}
    23.       ]}
    24.     ]},
    26.     %% 'access' parameter: sub = 1, pub = 2
    28.     {acl_req, [
    29.       {method, post},
    30.       {url, "http://localhost:8080/mqtt/acl"},
    31.       {params, [
    32.         {access,   "%A"},
    33.         {username, "%u"},
    34.         {clientid, "%c"},
    35.         {ipaddr,   "%a"},
    36.         {topic,    "%t"}
    37.       ]}
    38.     ]}
    39.   ]}
    41. ].

    HTTP API

    认证/ACL成功，API返回200

    认证/ACL失败，API返回4xx

    加载插件

    ./bin/emqttd_ctl plugins load emqttd_auth_http

    emqttd_plugin_mysql: MySQL认证/访问控制插件

    MySQL认证/访问控制插件，基于MySQL库表认证鉴权: https://github.com/emqtt/emqttd_plugin_mysql

    MQTT用户表

    1. CREATE TABLE `mqtt_user` (
    2.   `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
    3.   `username` varchar(100) DEFAULT NULL,
    4.   `password` varchar(100) DEFAULT NULL,
    5.   `salt` varchar(20) DEFAULT NULL,
    6.   `is_superuser` tinyint(1) DEFAULT 0,
    7.   `created` datetime DEFAULT NULL,
    8.   PRIMARY KEY (`id`),
    9.   UNIQUE KEY `mqtt_username` (`username`)
    10. ) ENGINE=MyISAM DEFAULT CHARSET=utf8;

    Note

    MQTT访问控制表

    1. CREATE TABLE `mqtt_acl` (
    2.   `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
    3.   `allow` int(1) DEFAULT NULL COMMENT '0: deny, 1: allow',
    4.   `ipaddr` varchar(60) DEFAULT NULL COMMENT 'IpAddress',
    5.   `username` varchar(100) DEFAULT NULL COMMENT 'Username',
    6.   `clientid` varchar(100) DEFAULT NULL COMMENT 'ClientId',
    7.   `access` int(2) NOT NULL COMMENT '1: subscribe, 2: publish, 3: pubsub',
    8.   `topic` varchar(100) NOT NULL DEFAULT '' COMMENT 'Topic Filter',
    9.   PRIMARY KEY (`id`)
    10. ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
    12. INSERT INTO `mqtt_acl` (`id`, `allow`, `ipaddr`, `username`, `clientid`, `access`, `topic`)
    13. VALUES
    14.     (1,1,NULL,'$all',NULL,2,'#'),
    15.     (2,0,NULL,'$all',NULL,1,'$SYS/#'),
    16.     (3,0,NULL,'$all',NULL,1,'eq #'),
    17.     (5,1,'127.0.0.1',NULL,NULL,2,'$SYS/#'),
    18.     (6,1,'127.0.0.1',NULL,NULL,2,'#'),
    19.     (7,1,NULL,'dashboard',NULL,1,'$SYS/#');

    配置插件

    plugins/emqttd_plugin_mysql/etc/plugin.config:

    1. [
    3.   {emqttd_plugin_mysql, [
    5.     {mysql_pool, [
    6.         %% ecpool options
    7.         {pool_size, 8},
    8.         {auto_reconnect, 3},
    10.         %% mysql options
    11.         {host,     "localhost"},
    12.         {port,     3306},
    13.         {user,     ""},
    14.         {password, ""},
    15.         {database, "mqtt"},
    16.         {encoding, utf8}
    17.     ]},
    19.     %% Variables: %u = username, %c = clientid, %a = ipaddress
    21.     %% Superuser Query
    22.     {superquery, "select is_superuser from mqtt_user where username = '%u' limit 1"},
    24.     {authquery, "select password from mqtt_user where username = '%u' limit 1"},
    26.     %% hash algorithm: plain, md5, sha, sha256, pbkdf2?
    27.     {password_hash, sha256},
    29.     %% select password with salt
    30.     %% {authquery, "select password, salt from mqtt_user where username = '%u'"},
    32.     %% sha256 with salt prefix
    33.     %% {password_hash, {salt, sha256}},
    35.     %% sha256 with salt suffix
    36.     %% {password_hash, {sha256, salt}},
    38.     %% '%a' = ipaddress, '%u' = username, '%c' = clientid
    39.     %% Comment this query, the acl will be disabled
    40.     {aclquery, "select allow, ipaddr, username, clientid, access, topic from mqtt_acl where ipaddr = '%a' or username = '%u' or username = '$all' or clientid = '%c'"},
    42.     %% If no ACL rules matched, return...
    43.     {acl_nomatch, allow}
    45.   ]}
    47. ].

    加载插件

    ./bin/emqttd_ctl plugins load emqttd_plugin_mysql

    PostgreSQL认证/访问控制插件，基于PostgreSQL库表认证鉴权:

    MQTT用户表

    1. CREATE TABLE mqtt_user (
    2.   id SERIAL primary key,
    3.   is_superuser boolean,
    4.   username character varying(100),
    5.   password character varying(100),
    6.   salt character varying(40)
    7. );

    MQTT访问控制表

    1. CREATE TABLE mqtt_acl (
    3.   allow integer,
    4.   ipaddr character varying(60),
    5.   username character varying(100),
    6.   clientid character varying(100),
    7.   access  integer,
    8.   topic character varying(100)
    9. );
    11. INSERT INTO mqtt_acl (id, allow, ipaddr, username, clientid, access, topic)
    12. VALUES
    13.     (1,1,NULL,'$all',NULL,2,'#'),
    14.     (2,0,NULL,'$all',NULL,1,'$SYS/#'),
    15.     (3,0,NULL,'$all',NULL,1,'eq #'),
    16.     (5,1,'127.0.0.1',NULL,NULL,2,'$SYS/#'),
    17.     (6,1,'127.0.0.1',NULL,NULL,2,'#'),
    18.     (7,1,NULL,'dashboard',NULL,1,'$SYS/#');

    配置插件

    plugins/emqttd_plugin_pgsql/etc/plugin.config:

    1. [
    3.   {emqttd_plugin_pgsql, [
    5.     {pgsql_pool, [
    6.         %% ecpool options
    7.         {pool_size, 8},
    8.         {auto_reconnect, 3},
    10.         %% pgsql options
    11.         {host, "localhost"},
    12.         {port, 5432},
    13.         {ssl, false},
    14.         {username, "feng"},
    15.         {password, ""},
    16.         {database, "mqtt"},
    17.         {encoding,  utf8}
    18.     ]},
    20.     %% Variables: %u = username, %c = clientid, %a = ipaddress
    22.     %% Superuser Query
    23.     {superquery, "select is_superuser from mqtt_user where username = '%u' limit 1"},
    25.     %% Authentication Query: select password only
    26.     {authquery, "select password from mqtt_user where username = '%u' limit 1"},
    28.     %% hash algorithm: plain, md5, sha, sha256, pbkdf2?
    29.     {password_hash, sha256},
    31.     %% select password with salt
    32.     %% {authquery, "select password, salt from mqtt_user where username = '%u'"},
    34.     %% sha256 with salt prefix
    35.     %% {password_hash, {salt, sha256}},
    37.     %% sha256 with salt suffix
    38.     %% {password_hash, {sha256, salt}},
    40.     %% Comment this query, the acl will be disabled. Notice: don't edit this query!
    41.     {aclquery, "select allow, ipaddr, username, clientid, access, topic from mqtt_acl
    42.                  where ipaddr = '%a' or username = '%u' or username = '$all' or clientid = '%c'"},
    44.     %% If no rules matched, return...
    45.     {acl_nomatch, allow}
    46.   ]}
    47. ].
    1. ./bin/emqttd_ctl plugins load emqttd_plugin_pgsql

    emqttd_plugin_redis: Redis认证/访问控制插件

    基于Redis认证/访问控制: https://github.com/emqtt/emqttd_plugin_redis

    配置插件

    plugins/emqttd_plugin_redis/etc/plugin.config:

    1. [
    2.   {emqttd_plugin_redis, [
    4.     {eredis_pool, [
    5.       %% ecpool options
    6.       {pool_size, 8},
    7.       {auto_reconnect, 2},
    9.       %% eredis options
    10.       {host, "127.0.0.1"},
    11.       {port, 6379},
    12.       {database, 0},
    13.       {password, ""}
    14.     ]},
    16.     %% Variables: %u = username, %c = clientid
    18.     %% HMGET mqtt_user:%u is_superuser
    19.     {supercmd, ["HGET", "mqtt_user:%u", "is_superuser"]},
    21.     %% HMGET mqtt_user:%u password
    22.     {authcmd, ["HGET", "mqtt_user:%u", "password"]},
    24.     %% Password hash algorithm: plain, md5, sha, sha256, pbkdf2?
    25.     {password_hash, sha256},
    27.     %% SMEMBERS mqtt_acl:%u
    28.     {aclcmd, ["SMEMBERS", "mqtt_acl:%u"]},
    30.     %% If no rules matched, return...
    31.     {acl_nomatch, deny},
    33.     %% Load Subscriptions form Redis when client connected.
    34.     {subcmd, ["HGETALL", "mqtt_subs:%u"]}
    35.   ]}
    36. ].

    用户Hash

    默认基于用户Hash认证:

    ACL规则SET

    默认采用SET存储ACL规则:

    1. SADD mqtt_acl:<username> "publish topic1"
    2. SADD mqtt_acl:<username> "subscribe topic2"
    3. SADD mqtt_acl:<username> "pubsub topic3"

    订阅Hash

    插件还支持Redis中创建MQTT订阅。当MQTT客户端连接成功，会自动从Redis加载订阅:

    1. HSET mqtt_subs:<username> topic1 0
    2. HSET mqtt_subs:<username> topic2 1
    3. HSET mqtt_subs:<username> topic3 2

    加载插件

    1. ./bin/emqttd_ctl plugins load emqttd_plugin_redis

    emqttd_plugin_mongo: MongoDB认证/访问控制插件

    基于MongoDB认证/访问控制:

    配置插件

    plugins/emqttd_plugin_mongo/etc/plugin.config:

    1. [
    2.   {emqttd_plugin_mongo, [
    4.     {mongo_pool, [
    5.       {pool_size, 8},
    6.       {auto_reconnect, 3},
    8.       %% Mongodb Driver Opts
    9.       {host, "localhost"},
    10.       {port, 27017},
    11.       %% {login, ""},
    12.       %% {password, ""},
    13.     ]},
    15.     %% Variables: %u = username, %c = clientid
    17.     %% Superuser Query
    18.     {superquery, [
    19.       {collection, "mqtt_user"},
    20.       {super_field, "is_superuser"},
    21.       {selector, {"username", "%u"}}
    22.     ]},
    24.     %% Authentication Query
    25.     {authquery, [
    26.       {collection, "mqtt_user"},
    27.       {password_field, "password"},
    28.       %% Hash Algorithm: plain, md5, sha, sha256, pbkdf2?
    30.       {selector, {"username", "%u"}}
    31.     ]},
    33.     %% ACL Query: "%u" = username, "%c" = clientid
    34.     {aclquery, [
    35.       {collection, "mqtt_acl"},
    36.       {selector, {"username", "%u"}}
    37.     ]},
    39.     %% If no ACL rules matched, return...
    40.     {acl_nomatch, deny}
    42.   ]}
    43. ].

    MongoDB数据库

    1. use mqtt
    2. db.createCollection("mqtt_user")
    3. db.createCollection("mqtt_acl")
    4. db.mqtt_user.ensureIndex({"username":1})

    Note

    数据库、集合名称可自定义

    用户集合(User Collection)

    1. {
    2.     username: "user",
    3.     password: "password hash",
    4.     is_superuser: boolean (true, false),
    5.     created: "datetime"
    6. }

    示例:

    1. db.mqtt_user.insert({username: "test", password: "password hash", is_superuser: false})
    2. db.mqtt_user:insert({username: "root", is_superuser: true})

    ACL集合(ACL Collection)

    1. {
    2.     username: "username",
    3.     clientid: "clientid",
    4.     publish: ["topic1", "topic2", ...],
    5.     subscribe: ["subtop1", "subtop2", ...],
    6.     pubsub: ["topic/#", "topic1", ...]
    7. }

    示例:

    1. db.mqtt_acl.insert({username: "test", publish: ["t/1", "t/2"], subscribe: ["user/%u", "client/%c"]})
    2. db.mqtt_acl.insert({username: "admin", pubsub: ["#"]})

    加载插件

    1. ./bin/emqttd_ctl plugins load emqttd_plugin_mongo

    emqttd_stomp: Stomp协议插件

    Stomp协议插件。支持STOMP 1.0/1.1/1.2协议客户端连接emqttd，发布订阅MQTT消息。

    配置插件

    Note

    Stomp协议端口: 61613

    plugins/emqttd_stomp/etc/plugin.config:

    1. [
    2.   {emqttd_stomp, [
    4.     {default_user, [
    5.         {login,    "guest"},
    6.         {passcode, "guest"}
    7.     ]},
    9.     {allow_anonymous, true},
    11.     %%TODO: unused...
    12.     {frame, [
    13.       {max_headers,       10},
    14.       {max_header_length, 1024},
    15.       {max_body_length,   8192}
    16.     ]},
    18.     {listeners, [
    19.       {emqttd_stomp, 61613, [
    20.         {acceptors,   4},
    21.         {max_clients, 512}
    22.       ]}
    23.     ]}
    25.   ]}
    26. ].
    1. ./bin/emqttd_ctl plugins load emqttd_stomp

    配置插件

    Note

    缺省端口: 61616

    加载插件

    Note

    需先加载emqttd_stomp插件

    1. ./bin/emqttd_ctl plugins load emqttd_stomp
    3. ./bin/emqttd_ctl plugins load emqttd_sockjs

    插件演示页面

    emqttd_recon: Recon性能调试插件

    emqttd_recon插件集成recon性能调测库，’./bin/emqttd_ctl’命令行注册recon命令。

    加载插件

    1. ./bin/emqttd_ctl plugins load emqttd_recon

    recon命令

    1. ./bin/emqttd_ctl recon
    3. recon memory                 #recon_alloc:memory/2
    4. recon allocated              #recon_alloc:memory(allocated_types, current|max)
    5. recon bin_leak               #recon:bin_leak(100)
    6. recon node_stats             #recon:node_stats(10, 1000)
    7. recon remote_load Mod        #recon:remote_load(Mod)

    emqttd_reloader: 代码热加载插件

    用于开发调试的代码热升级插件。加载该插件后，emqttd会自动热升级更新代码。

    Note

    加载插件

    1. ./bin/emqttd_ctl plugins load emqttd_reloader

    reload命令

    1. ./bin/emqttd_ctl reload
    3. reload <Module>             # Reload a Module

    emqttd插件开发

    创建插件项目

    github下载emqttd源码库，plugins/目录下创建插件应用。

    模版代码请参考: emqttd_plugin_templage

    注册认证/访问控制模块

    认证演示模块 - emqttd_auth_demo.erl

    1. -module(emqttd_auth_demo).
    3. -behaviour(emqttd_auth_mod).
    5. -include("../../../include/emqttd.hrl").
    7. -export([init/1, check/3, description/0]).
    9. init(Opts) -> {ok, Opts}.
    11. check(#mqtt_client{client_id = ClientId, username = Username}, Password, _Opts) ->
    12.     io:format("Auth Demo: clientId=~p, username=~p, password=~p~n",
    13.               [ClientId, Username, Password]),
    14.     ok.
    16. description() -> "Demo Auth Module".

    访问控制演示模块 - emqttd_acl_demo.erl

    1. -module(emqttd_acl_demo).
    3. -include("../../../include/emqttd.hrl").
    5. %% ACL callbacks
    6. -export([init/1, check_acl/2, reload_acl/1, description/0]).
    8. init(Opts) ->
    9.     {ok, Opts}.
    11. check_acl({Client, PubSub, Topic}, Opts) ->
    12.     io:format("ACL Demo: ~p ~p ~p~n", [Client, PubSub, Topic]),
    13.     allow.
    15. reload_acl(_Opts) ->
    16.     ok.
    18. description() -> "ACL Module Demo".

    注册认证、访问控制模块 - emqttd_plugin_template_app.erl

    1. ok = emqttd_access_control:register_mod(auth, emqttd_auth_demo, []),
    2. ok = emqttd_access_control:register_mod(acl, emqttd_acl_demo, []),

    注册扩展钩子(Hooks)

    通过钩子(Hook)处理客户端上下线、主题订阅、消息收发。

    emqttd_plugin_template.erl:

    1. %% Called when the plugin application start
    2. load(Env) ->
    3.     emqttd:hook('client.connected', fun ?MODULE:on_client_connected/3, [Env]),
    4.     emqttd:hook('client.disconnected', fun ?MODULE:on_client_disconnected/3, [Env]),
    5.     emqttd:hook('client.subscribe', fun ?MODULE:on_client_subscribe/3, [Env]),
    6.     emqttd:hook('client.subscribe.after', fun ?MODULE:on_client_subscribe_after/3, [Env]),
    7.     emqttd:hook('client.unsubscribe', fun ?MODULE:on_client_unsubscribe/3, [Env]),
    8.     emqttd:hook('message.publish', fun ?MODULE:on_message_publish/2, [Env]),
    9.     emqttd:hook('message.delivered', fun ?MODULE:on_message_delivered/3, [Env]),
    10.     emqttd:hook('message.acked', fun ?MODULE:on_message_acked/3, [Env]).

    扩展钩子(Hook):

    钩子

    说明

    client.connected

    客户端上线

    client.subscribe

    客户端订阅主题前

    client.subscribe.after

    客户端订阅主题后

    client.unsubscribe

    客户端取消订阅主题

    message.publish

    MQTT消息发布

    message.delivered

    MQTT消息送达

    message.acked

    MQTT消息回执

    client.disconnected

    客户端连接断开

    注册扩展命令行

    扩展命令行演示模块 - emqttd_cli_demo.erl

    1. -module(emqttd_cli_demo).
    3. -include("../../../include/emqttd_cli.hrl").
    5. -export([cmd/1]).
    7. cmd(["arg1", "arg2"]) ->
    8.     ?PRINT_MSG("ok");
    10. cmd(_) ->
    11.     ?USAGE([{"cmd arg1 arg2",  "cmd demo"}]).

      插件加载后，’./bin/emqttd_ctl’新增命令行: