扩展插件(Plugins)

emqttd消息服务器通过模块注册和钩子(Hooks)机制，支持用户开发扩展插件定制服务器认证鉴权与业务功能。

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

emqttd_plugin_template: 插件开发模版

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

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

目录/文件

说明

etc/plugin.config

插件配置文件

ebin/

插件程序文件目录

加载、卸载插件

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

加载插件:

  1. ./bin/emqttd_ctl plugins load <PluginName>

卸载插件:

  1. ./bin/emqttd_ctl plugins unload <PluginName>

查询插件:

  1. ./bin/emqttd_ctl plugins list

emqttd_dashboard: Dashboard插件

emqttd消息服务器的Web管理控制台。插件项目地址: https://github.com/emqtt/emqttd_dashboard

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

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

_images/dashboard.png

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认证/访问控制插件: https://github.com/emqtt/emqttd_auth_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

MySQL插件可使用系统自有的用户表，通过’authquery’配置查询语句。

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.     %% Authentication Query: select password only
  25.     {authquery, "select password from mqtt_user where username = '%u' limit 1"},
  27.     %% hash algorithm: plain, md5, sha, sha256, pbkdf2?
  28.     {password_hash, sha256},
  30.     %% select password with salt
  31.     %% {authquery, "select password, salt from mqtt_user where username = '%u'"},
  33.     %% sha256 with salt prefix
  34.     %% {password_hash, {salt, sha256}},
  36.     %% sha256 with salt suffix
  37.     %% {password_hash, {sha256, salt}},
  39.     %% '%a' = ipaddress, '%u' = username, '%c' = clientid
  40.     %% Comment this query, the acl will be disabled
  41.     {aclquery, "select allow, ipaddr, username, clientid, access, topic from mqtt_acl where ipaddr = '%a' or username = '%u' or username = '$all' or clientid = '%c'"},
  43.     %% If no ACL rules matched, return...
  44.     {acl_nomatch, allow}
  46.   ]}
  48. ].

加载插件

./bin/emqttd_ctl plugins load emqttd_plugin_mysql

emqttd_plugin_pgsql: PostgreSQL认证/访问控制插件

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

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 (
  2.   id SERIAL primary key,
  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认证:

  1. HSET mqtt_user:<username> is_superuser 1
  2. HSET mqtt_user:<username> password "passwd"

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认证/访问控制: https://github.com/emqtt/emqttd_plugin_mongo

配置插件

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.       {database, "mqtt"}
  14.     ]},
  16.     %% Variables: %u = username, %c = clientid
  18.     %% Superuser Query
  19.     {superquery, [
  20.       {collection, "mqtt_user"},
  21.       {super_field, "is_superuser"},
  22.       {selector, {"username", "%u"}}
  23.     ]},
  25.     %% Authentication Query
  26.     {authquery, [
  27.       {collection, "mqtt_user"},
  28.       {password_field, "password"},
  29.       %% Hash Algorithm: plain, md5, sha, sha256, pbkdf2?
  30.       {password_hash, sha256},
  31.       {selector, {"username", "%u"}}
  32.     ]},
  34.     %% ACL Query: "%u" = username, "%c" = clientid
  35.     {aclquery, [
  36.       {collection, "mqtt_acl"},
  37.       {selector, {"username", "%u"}}
  38.     ]},
  40.     %% If no ACL rules matched, return...
  41.     {acl_nomatch, deny}
  43.   ]}
  44. ].

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

emqttd_sockjs: Stomp/Sockjs插件

配置插件

Note

缺省端口: 61616

  1. [
  2.   {emqttd_sockjs, [
  4.     {sockjs, []},
  6.     {cowboy_listener, {stomp_sockjs, 61616, 4}},
  8.   ]}
  9. ].

加载插件

Note

需先加载emqttd_stomp插件

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

插件演示页面

http://localhost:61616/index.html

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"}]).

注册命令行模块 - emqttd_plugin_template_app.erl

  1. emqttd_ctl:register_cmd(cmd, {emqttd_cli_demo, cmd}, []).

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

  1. ./bin/emqttd_ctl cmd arg1 arg2