diff --git a/test/emqttd_SUITE.erl b/test/emqttd_SUITE.erl index d7af619df..e78d856b1 100644 --- a/test/emqttd_SUITE.erl +++ b/test/emqttd_SUITE.erl @@ -47,7 +47,8 @@ all() -> {group, http}, {group, cluster}, {group, alarms}, - {group, cli}]. + {group, cli}, + {group, cleanSession}]. groups() -> [{protocol, [sequence], @@ -103,7 +104,11 @@ groups() -> cli_bridges, cli_plugins, cli_listeners, - cli_vm]}]. + cli_vm]}, + {cleanSession, [sequence], + [cleanSession_validate, + cleanSession_validate1, + cleanSession_validate2]}]. init_per_suite(Config) -> application:start(lager), @@ -616,6 +621,59 @@ cli_vm(_) -> emqttd_cli:vm([]), emqttd_cli:vm(["ports"]). +cleanSession_validate(_) -> + {ok, C1} = emqttc:start_link([{host, "localhost"}, + {port, 1883}, + {client_id, <<"c1">>}, + {clean_sess, false}]), + timer:sleep(10), + emqttc:subscribe(C1, <<"topic">>, qos0), + ok = emqttd_cli:sessions(["list", "persistent"]), + emqttc:disconnect(C1), + {ok, Pub} = emqttc:start_link([{host, "localhost"}, + {port, 1883}, + {client_id, <<"pub">>}]), + + emqttc:publish(Pub, <<"topic">>, <<"m1">>, [{qos, 0}]), + timer:sleep(10), + {ok, C11} = emqttc:start_link([{host, "localhost"}, + {port, 1883}, + {client_id, <<"c1">>}, + {clean_sess, false}]), + timer:sleep(100), + Metrics = emqttd_metrics:all(), + ct:log("Metrics:~p~n", [Metrics]), + ?assertEqual(1, proplists:get_value('messages/qos0/sent', Metrics)), + ?assertEqual(1, proplists:get_value('messages/qos0/received', Metrics)), + emqttc:disconnect(Pub), + emqttc:disconnect(C11). + +cleanSession_validate1(_) -> + {ok, C1} = emqttc:start_link([{host, "localhost"}, + {port, 1883}, + {client_id, <<"c1">>}, + {clean_sess, true}]), + timer:sleep(10), + emqttc:subscribe(C1, <<"topic">>, qos1), + ok = emqttd_cli:sessions(["list", "transient"]), + emqttc:disconnect(C1), + {ok, Pub} = emqttc:start_link([{host, "localhost"}, + {port, 1883}, + {client_id, <<"pub">>}]), + + emqttc:publish(Pub, <<"topic">>, <<"m1">>, [{qos, 1}]), + timer:sleep(10), + {ok, C11} = emqttc:start_link([{host, "localhost"}, + {port, 1883}, + {client_id, <<"c1">>}, + {clean_sess, false}]), + timer:sleep(100), + Metrics = emqttd_metrics:all(), + ?assertEqual(0, proplists:get_value('messages/qos1/sent', Metrics)), + ?assertEqual(1, proplists:get_value('messages/qos1/received', Metrics)), + emqttc:disconnect(Pub), + emqttc:disconnect(C11). + ensure_ok(ok) -> ok; ensure_ok({error, {already_started, _}}) -> ok. diff --git a/test/emqttd_SUITE_data/emqttd.conf b/test/emqttd_SUITE_data/emqttd.conf index 4aaa60eff..a15aeada9 100644 --- a/test/emqttd_SUITE_data/emqttd.conf +++ b/test/emqttd_SUITE_data/emqttd.conf @@ -1,3 +1,8 @@ + +##=================================================================== +## EMQ Configuration R2.1 +##=================================================================== + ##-------------------------------------------------------------------- ## Node Args ##-------------------------------------------------------------------- @@ -11,6 +16,11 @@ node.cookie = emq_dist_cookie ## SMP support: enable, auto, disable node.smp = auto +## vm.args: -heart +## Heartbeat monitoring of an Erlang runtime system +## Value should be 'on' or comment the line +## node.heartbeat = on + ## Enable kernel poll node.kernel_poll = on @@ -40,21 +50,30 @@ node.crash_dump = log/crash.dump node.dist_net_ticktime = 60 ## Distributed node port range -## node.dist_listen_min = 6000 -## node.dist_listen_max = 6999 +## node.dist_listen_min = 6369 +## node.dist_listen_max = 6369 ##-------------------------------------------------------------------- ## Log ##-------------------------------------------------------------------- +## Set the log dir +log.dir = {{ platform_log_dir }} + ## Console log. Enum: off, file, console, both log.console = console +## Syslog. Enum: on, off +log.syslog = on + +## syslog level. Enum: debug, info, notice, warning, error, critical, alert, emergency +log.syslog.level = error + ## Console log level. Enum: debug, info, notice, warning, error, critical, alert, emergency log.console.level = error ## Console log file -## log.console.file = log/console.log +## log.console.file = {{ platform_log_dir }}/console.log ## Error log file log.error.file = log/error.log @@ -64,6 +83,19 @@ log.crash = on log.crash.file = log/crash.log +##-------------------------------------------------------------------- +## Allow Anonymous and Default ACL +##-------------------------------------------------------------------- + +## Allow Anonymous authentication +mqtt.allow_anonymous = true + +## Default ACL File +mqtt.acl_file = etc/acl.conf + +## Cache ACL for PUBLISH +mqtt.cache_acl = true + ##-------------------------------------------------------------------- ## MQTT Protocol ##-------------------------------------------------------------------- @@ -74,34 +106,38 @@ mqtt.max_clientid_len = 1024 ## Max Packet Size Allowed, 64K by default. mqtt.max_packet_size = 64KB +##-------------------------------------------------------------------- +## MQTT Client +##-------------------------------------------------------------------- + ## Client Idle Timeout (Second) -mqtt.client_idle_timeout = 30 +mqtt.client.idle_timeout = 30s -## Allow Anonymous authentication -mqtt.allow_anonymous = true - -## Default ACL File -mqtt.acl_file = etc/acl.conf +## Enable client Stats: seconds or off +mqtt.client.enable_stats = off ##-------------------------------------------------------------------- ## MQTT Session ##-------------------------------------------------------------------- +## Upgrade QoS? +mqtt.session.upgrade_qos = off + ## Max number of QoS 1 and 2 messages that can be “inflight” at one time. ## 0 means no limit -mqtt.session.max_inflight = 100 +mqtt.session.max_inflight = 32 -## Retry interval for redelivering QoS1/2 messages. -mqtt.session.retry_interval = 60 - -## Awaiting PUBREL Timeout -mqtt.session.await_rel_timeout = 20 +## Retry Interval for redelivering QoS1/2 messages. +mqtt.session.retry_interval = 20s ## Max Packets that Awaiting PUBREL, 0 means no limit -mqtt.session.max_awaiting_rel = 0 +mqtt.session.max_awaiting_rel = 100 -## Statistics Collection Interval(seconds) -mqtt.session.collect_interval = 0 +## Awaiting PUBREL Timeout +mqtt.session.await_rel_timeout = 20s + +## Enable Statistics at the Interval(seconds) +mqtt.session.enable_stats = off ## Expired after 1 day: ## w - week @@ -109,7 +145,7 @@ mqtt.session.collect_interval = 0 ## h - hour ## m - minute ## s - second -mqtt.session.expired_after = 1d +mqtt.session.expiry_interval = 2h ##-------------------------------------------------------------------- ## MQTT Queue @@ -204,12 +240,13 @@ mqtt.listener.ssl.max_clients = 512 ## Rate Limit. Format is 'burst,rate', Unit is KB/Sec ## mqtt.listener.ssl.rate_limit = 100,10 -## Configuring SSL Options -## See http://erlang.org/doc/man/ssl.html -mqtt.listener.ssl.handshake_timeout = 15 +## Configuring SSL Options. See http://erlang.org/doc/man/ssl.html +### TLS only for POODLE attack +mqtt.listener.ssl.tls_versions = tlsv1.2,tlsv1.1,tlsv1 +mqtt.listener.ssl.handshake_timeout = 15s mqtt.listener.ssl.keyfile = certs/key.pem mqtt.listener.ssl.certfile = certs/cert.pem -## mqtt.listener.ssl.cacertfile = etc/certs/cacert.pem +## mqtt.listener.ssl.cacertfile = {{ platform_etc_dir }}/certs/cacert.pem ## mqtt.listener.ssl.verify = verify_peer ## mqtt.listener.ssl.fail_if_no_peer_cert = true @@ -219,13 +256,14 @@ mqtt.listener.http.acceptors = 4 mqtt.listener.http.max_clients = 64 ## HTTP(SSL) Listener -## mqtt.listener.https = 8084 -## mqtt.listener.https.acceptors = 4 -## mqtt.listener.https.max_clients = 64 -## mqtt.listener.https.handshake_timeout = 15 -## mqtt.listener.https.certfile = etc/certs/cert.pem -## mqtt.listener.https.keyfile = etc/certs/key.pem -## mqtt.listener.https.cacertfile = etc/certs/cacert.pem +mqtt.listener.https = 8084 +mqtt.listener.https.acceptors = 4 +mqtt.listener.https.max_clients = 64 +mqtt.listener.https.handshake_timeout = 15 +mqtt.listener.https.keyfile = certs/key.pem +mqtt.listener.https.certfile = certs/cert.pem +## mqtt.listener.https.cacertfile = {{ platform_etc_dir }}/certs/cacert.pem + ## mqtt.listener.https.verify = verify_peer ## mqtt.listener.https.fail_if_no_peer_cert = true diff --git a/test/emqttd_SUITE_data/emqttd.schema b/test/emqttd_SUITE_data/emqttd.schema index 2c73482ed..156bc4874 100644 --- a/test/emqttd_SUITE_data/emqttd.schema +++ b/test/emqttd_SUITE_data/emqttd.schema @@ -22,6 +22,19 @@ hidden ]}. +%% @doc http://erlang.org/doc/man/heart.html +{mapping, "node.heartbeat", "vm_args.-heart", [ + {datatype, flag}, + hidden +]}. + +{translation, "vm_args.-heart", fun(Conf) -> + case cuttlefish:conf_get("node.heartbeat", Conf) of + true -> ""; + false -> cuttlefish:invalid("should be 'on' or comment the line!") + end +end}. + %% @doc Enable Kernel Poll {mapping, "node.kernel_poll", "vm_args.+K", [ {default, on}, @@ -135,8 +148,13 @@ %% Log %%-------------------------------------------------------------------- +{mapping, "log.dir", "lager.log_dir", [ + {default, "log"}, + {datatype, string} +]}. + {mapping, "log.console", "lager.handlers", [ - {default, file }, + {default, file}, {datatype, {enum, [off, file, console, both]}} ]}. @@ -155,6 +173,26 @@ {datatype, file} ]}. +{mapping, "log.syslog", "lager.handlers", [ + {default, off}, + {datatype, flag} +]}. + +{mapping, "log.syslog.identity", "lager.handlers", [ + {default, "emq"}, + {datatype, string} +]}. + +{mapping, "log.syslog.facility", "lager.handlers", [ + {default, local0}, + {datatype, {enum, [daemon, local0, local1, local2, local3, local4, local5, local6, local7]}} +]}. + +{mapping, "log.syslog.level", "lager.handlers", [ + {default, err}, + {datatype, {enum, [debug, info, notice, warning, error, critical, alert, emergency]}} +]}. + {mapping, "log.error.redirect", "lager.error_logger_redirect", [ {default, on}, {datatype, flag}, @@ -196,7 +234,16 @@ both -> [ConsoleHandler, ConsoleFileHandler]; _ -> [] end, - ConsoleHandlers ++ ErrorHandler + + SyslogHandler = case cuttlefish:conf_get("log.syslog", Conf) of + false -> []; + true -> [{lager_syslog_backend, + [cuttlefish:conf_get("log.syslog.identity", Conf), + cuttlefish:conf_get("log.syslog.facility", Conf), + cuttlefish:conf_get("log.syslog.level", Conf)]}] + end, + + ConsoleHandlers ++ ErrorHandler ++ SyslogHandler end }. @@ -226,6 +273,28 @@ hidden ]}. +%%-------------------------------------------------------------------- +%% Allow Anonymous and Default ACL +%%-------------------------------------------------------------------- + +%% @doc Allow Anonymous +{mapping, "mqtt.allow_anonymous", "emqttd.allow_anonymous", [ + {default, false}, + {datatype, {enum, [true, false]}} +]}. + +%% @doc Default ACL File +{mapping, "mqtt.acl_file", "emqttd.acl_file", [ + {datatype, string}, + hidden +]}. + +%% @doc Cache ACL for PUBLISH +{mapping, "mqtt.cache_acl", "emqttd.cache_acl", [ + {default, true}, + {datatype, {enum, [true, false]}} +]}. + %%-------------------------------------------------------------------- %% MQTT Protocol %%-------------------------------------------------------------------- @@ -242,35 +311,42 @@ {datatype, bytesize} ]}. -%% @doc Client Idle Timeout. -{mapping, "mqtt.client_idle_timeout", "emqttd.protocol", [ - {default, 30}, - {datatype, integer} -]}. - {translation, "emqttd.protocol", fun(Conf) -> - [{max_clientid_len, cuttlefish:conf_get("mqtt.max_clientid_len", Conf)}, - {max_packet_size, cuttlefish:conf_get("mqtt.max_packet_size", Conf)}, - {client_idle_timeout, cuttlefish:conf_get("mqtt.client_idle_timeout", Conf)}] + [{max_clientid_len, cuttlefish:conf_get("mqtt.max_clientid_len", Conf)}, + {max_packet_size, cuttlefish:conf_get("mqtt.max_packet_size", Conf)}] end}. -%% @doc Allow Anonymous -{mapping, "mqtt.allow_anonymous", "emqttd.allow_anonymous", [ - {default, false}, - {datatype, {enum, [true, false]}}, - hidden +%%-------------------------------------------------------------------- +%% MQTT Client +%%-------------------------------------------------------------------- + +%% @doc Client Idle Timeout. +{mapping, "mqtt.client.idle_timeout", "emqttd.client", [ + {default, "30s"}, + {datatype, {duration, ms}} ]}. -%% @doc Default ACL File -{mapping, "mqtt.acl_file", "emqttd.acl_file", [ - {datatype, string}, - hidden +%% @doc Enable Stats of Client. +{mapping, "mqtt.client.enable_stats", "emqttd.client", [ + {default, off}, + {datatype, [{duration, ms}, flag]} ]}. +%% @doc Client +{translation, "emqttd.client", fun(Conf) -> + [{client_idle_timeout, cuttlefish:conf_get("mqtt.client.idle_timeout", Conf)}, + {client_enable_stats, cuttlefish:conf_get("mqtt.client.enable_stats", Conf)}] +end}. + %%-------------------------------------------------------------------- %% MQTT Session %%-------------------------------------------------------------------- +%% @doc Upgrade QoS? +{mapping, "mqtt.session.upgrade_qos", "emqttd.session", [ + {default, off}, + {datatype, flag} +]}. %% @doc Max number of QoS 1 and 2 messages that can be “inflight” at one time. %% 0 means no limit {mapping, "mqtt.session.max_inflight", "emqttd.session", [ @@ -278,17 +354,10 @@ end}. {datatype, integer} ]}. - %% @doc Retry interval for redelivering QoS1/2 messages. {mapping, "mqtt.session.retry_interval", "emqttd.session", [ - {default, 60}, - {datatype, integer} -]}. - -%% @doc Awaiting PUBREL Timeout -{mapping, "mqtt.session.await_rel_timeout", "emqttd.session", [ - {default, 30}, - {datatype, integer} + {default, "20s"}, + {datatype, {duration, ms}} ]}. %% @doc Max Packets that Awaiting PUBREL, 0 means no limit @@ -297,25 +366,32 @@ end}. {datatype, integer} ]}. -%% @doc Statistics Collection Interval(seconds) -{mapping, "mqtt.session.collect_interval", "emqttd.session", [ - {default, 0}, - {datatype, integer} +%% @doc Awaiting PUBREL Timeout +{mapping, "mqtt.session.await_rel_timeout", "emqttd.session", [ + {default, "20s"}, + {datatype, {duration, ms}} ]}. -%% @doc Session expired after... -{mapping, "mqtt.session.expired_after", "emqttd.session", [ - {default, "2d"}, - {datatype, {duration, s}} +%% @doc Enable Stats +{mapping, "mqtt.session.enable_stats", "emqttd.session", [ + {default, off}, + {datatype, [{duration, ms}, flag]} +]}. + +%% @doc Session Expiry Interval +{mapping, "mqtt.session.expiry_interval", "emqttd.session", [ + {default, "2h"}, + {datatype, {duration, ms}} ]}. {translation, "emqttd.session", fun(Conf) -> - [{max_inflight, cuttlefish:conf_get("mqtt.session.max_inflight", Conf)}, - {retry_interval, cuttlefish:conf_get("mqtt.session.retry_interval", Conf)}, + [{upgrade_qos, cuttlefish:conf_get("mqtt.session.upgrade_qos", Conf)}, + {max_inflight, cuttlefish:conf_get("mqtt.session.max_inflight", Conf)}, + {retry_interval, cuttlefish:conf_get("mqtt.session.retry_interval", Conf)}, + {max_awaiting_rel, cuttlefish:conf_get("mqtt.session.max_awaiting_rel", Conf)}, {await_rel_timeout, cuttlefish:conf_get("mqtt.session.await_rel_timeout", Conf)}, - {max_awaiting_rel, cuttlefish:conf_get("mqtt.session.max_awaiting_rel", Conf)}, - {collect_interval, cuttlefish:conf_get("mqtt.session.collect_interval", Conf)}, - {expired_after, cuttlefish:conf_get("mqtt.session.expired_after", Conf)}] + {enable_stats, cuttlefish:conf_get("mqtt.session.enable_stats", Conf)}, + {expiry_interval, cuttlefish:conf_get("mqtt.session.expiry_interval", Conf)}] end}. %%-------------------------------------------------------------------- @@ -331,28 +407,25 @@ end}. %% @doc Topic Priority: 0~255, Default is 0 {mapping, "mqtt.queue.priority", "emqttd.queue", [ {default, ""}, - {datatype, string}, - hidden + {datatype, string} ]}. %% @doc Max queue length. Enqueued messages when persistent client disconnected, or inflight window is full. {mapping, "mqtt.queue.max_length", "emqttd.queue", [ {default, infinity}, - {datatype, [atom, integer]} + {datatype, [integer, {atom, infinity}]} ]}. %% @doc Low-water mark of queued messages {mapping, "mqtt.queue.low_watermark", "emqttd.queue", [ {default, "20%"}, - {datatype, string}, - hidden + {datatype, string} ]}. %% @doc High-water mark of queued messages {mapping, "mqtt.queue.high_watermark", "emqttd.queue", [ {default, "60%"}, - {datatype, string}, - hidden + {datatype, string} ]}. %% @doc Queue Qos0 messages? @@ -405,8 +478,7 @@ end}. {mapping, "mqtt.pubsub.async", "emqttd.pubsub", [ {default, true}, - {datatype, {enum, [true, false]}}, - hidden + {datatype, {enum, [true, false]}} ]}. {translation, "emqttd.pubsub", fun(Conf) -> @@ -451,7 +523,7 @@ end}. %%-------------------------------------------------------------------- {mapping, "mqtt.listener.tcp", "emqttd.listeners", [ - {default, 1883}, + %% {default, 1883}, {datatype, [integer, ip]} ]}. @@ -467,8 +539,7 @@ end}. {mapping, "mqtt.listener.tcp.rate_limit", "emqttd.listeners", [ {default, undefined}, - {datatype, string}, - hidden + {datatype, string} ]}. {mapping, "mqtt.listener.tcp.backlog", "emqttd.listeners", [ @@ -497,7 +568,7 @@ end}. ]}. {mapping, "mqtt.listener.ssl", "emqttd.listeners", [ - {default, 8883}, + %% {default, 8883}, {datatype, [integer, ip]} ]}. @@ -515,9 +586,13 @@ end}. {datatype, string} ]}. +{mapping, "mqtt.listener.ssl.tls_versions", "emqttd.listeners", [ + {datatype, string} +]}. + {mapping, "mqtt.listener.ssl.handshake_timeout", "emqttd.listeners", [ - {default, 15}, - {datatype, integer} + {default, "15s"}, + {datatype, {duration, ms}} ]}. {mapping, "mqtt.listener.ssl.keyfile", "emqttd.listeners", [ @@ -541,7 +616,7 @@ end}. ]}. {mapping, "mqtt.listener.http", "emqttd.listeners", [ - {default, 8883}, + %% {default, 8083}, {datatype, [integer, ip]} ]}. @@ -556,9 +631,8 @@ end}. ]}. {mapping, "mqtt.listener.https", "emqttd.listeners", [ - {default, undefined}, - {datatype, [integer, ip]}, - hidden + %%{default, 8084}, + {datatype, [integer, ip]} ]}. {mapping, "mqtt.listener.https.acceptors", "emqttd.listeners", [ @@ -610,8 +684,16 @@ end}. {buffer, cuttlefish:conf_get(Prefix ++ ".buffer", Conf, undefined)}, {nodelay, cuttlefish:conf_get(Prefix ++ ".nodelay", Conf, true)}]) end, + + SplitFun = fun(undefined) -> undefined; (S) -> string:tokens(S, ",") end, + SslOpts = fun(Prefix) -> - Filter([{handshake_timeout, cuttlefish:conf_get(Prefix ++ ".handshake_timeout", Conf) * 1000}, + Versions = case SplitFun(cuttlefish:conf_get(Prefix ++ ".tls_versions", Conf, undefined)) of + undefined -> undefined; + L -> [list_to_atom(V) || V <- L] + end, + Filter([{versions, Versions}, + {handshake_timeout, cuttlefish:conf_get(Prefix ++ ".handshake_timeout", Conf), undefined}, {keyfile, cuttlefish:conf_get(Prefix ++ ".keyfile", Conf, undefined)}, {certfile, cuttlefish:conf_get(Prefix ++ ".certfile", Conf, undefined)}, {cacertfile, cuttlefish:conf_get(Prefix ++ ".cacertfile", Conf, undefined)},