yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'release-v50' into master

This commit is contained in:
Zaiming (Stone) Shi 2022-10-24 11:46:31 +02:00
parent 918c011e33 11daaa08e2
commit 508c0e84d0
26 changed files with 1016 additions and 275 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

210
CHANGES-5.0.md
View File

@ -1,210 +0,0 @@
# 5.0.9
## Enhancements
* Add `cert_common_name` and `cert_subject` placeholder support for authz_http and authz_mongo.[#8973](https://github.com/emqx/emqx/pull/8973)
* Use milliseconds internally in emqx_delayed to store the publish time, improving precision.[#9060](https://github.com/emqx/emqx/pull/9060)
* More rigorous checking of flapping to improve stability of the system. [#9136](https://github.com/emqx/emqx/pull/9136)
* No message(s) echo for the message publish APIs [#9155](https://github.com/emqx/emqx/pull/9155)
Prior to this fix, the message publish APIs (`api/v5/publish` and `api/v5/publish/bulk`) echos the message back to the client in HTTP body.
This change fixed it to only send back the message ID.
* Add /trace/:name/log_detail HTTP API to return trace file's size and mtime [#9152](https://github.com/emqx/emqx/pull/9152)
* Allow clear retained/delayed data when client is banned.[#9139](https://github.com/emqx/emqx/pull/9139)
* Update `gen_rpc` library to version 3.0 [#9187](https://github.com/emqx/emqx/pull/9187)
## Bug fixes
* Check ACLs for last will testament topic before publishing the message. [#8930](https://github.com/emqx/emqx/pull/8930)
* Fix GET /listeners API crash When some nodes still in initial configuration. [#9002](https://github.com/emqx/emqx/pull/9002)
* Fix empty variable interpolation in authentication and authorization. Placeholders for undefined variables are rendered now as empty strings and do not cause errors anymore. [#8963](https://github.com/emqx/emqx/pull/8963)
* Fix the latency statistics error of the slow subscription module when `stats_type` is `internal` or `response`. [#8986](https://github.com/emqx/emqx/pull/8986)
* Redispatch shared subscription messages. [#9104](https://github.com/emqx/emqx/pull/9104)
* Ensure authentication type is an array, not struct. [#8923](https://github.com/emqx/emqx/pull/8923)
# 5.0.8
## Bug fixes
* Fix exhook `client.authorize` never being execauted. [#8780](https://github.com/emqx/emqx/pull/8780)
* Fix JWT plugin don't support non-integer timestamp claims. [#8867](https://github.com/emqx/emqx/pull/8867)
* Avoid publishing will message when client fails to auhtenticate. [#8887](https://github.com/emqx/emqx/pull/8887)
* Speed up dispatching of shared subscription messages in a cluster [#8893](https://github.com/emqx/emqx/pull/8893)
* Fix the extra / prefix when CoAP gateway parsing client topics. [#8658](https://github.com/emqx/emqx/pull/8658)
* Speed up updating the configuration, When some nodes in the cluster are down. [#8857](https://github.com/emqx/emqx/pull/8857)
* Fix delayed publish inaccurate caused by os time change. [#8926](https://github.com/emqx/emqx/pull/8926)
* Fix that EMQX can't start when the retainer is disabled [#8911](https://github.com/emqx/emqx/pull/8911)
* Fix that redis authn will deny the unknown users [#8934](https://github.com/emqx/emqx/pull/8934)
* Fix ExProto UDP client keepalive checking error.
This causes the clients to not expire as long as a new UDP packet arrives [#8866](https://github.com/emqx/emqx/pull/8866)
* Fix that MQTT Bridge message payload could be empty string. [#8949](https://github.com/emqx/emqx/pull/8949)
## Enhancements
* Print a warning message when boot with the default (insecure) Erlang cookie. [#8905](https://github.com/emqx/emqx/pull/8905)
* Change the `/gateway` API path to plural form. [#8823](https://github.com/emqx/emqx/pull/8823)
* Don't allow updating config items when they already exist in `local-override.conf`. [#8851](https://github.com/emqx/emqx/pull/8851)
* Remove `node.etc_dir` from emqx.conf, because it is never used.
Also allow user to customize the logging directory [#8892](https://github.com/emqx/emqx/pull/8892)
* Added a new API `POST /listeners` for creating listener. [#8876](https://github.com/emqx/emqx/pull/8876)
* Close ExProto client process immediately if it's keepalive timeouted. [#8866](https://github.com/emqx/emqx/pull/8866)
* Upgrade grpc-erl driver to 0.6.7 to support batch operation in sending stream. [#8866](https://github.com/emqx/emqx/pull/8866)
# 5.0.7
## Bug fixes
* Remove `will_msg` (not used) field from the client API. [#8721](https://github.com/emqx/emqx/pull/8721)
* Fix `$queue` topic name error in management API return. [#8728](https://github.com/emqx/emqx/pull/8728)
* Fix race condition which may cause `client.connected` and `client.disconnected` out of order. [#8625](https://github.com/emqx/emqx/pull/8625)
* Fix quic listener default idle timeout's type. [#8826](https://github.com/emqx/emqx/pull/8826)
## Enhancements
* Do not auto-populate default SSL cipher suites, so that the configs are less bloated. [#8769](https://github.com/emqx/emqx/pull/8769)
# 5.0.6
## Bug fixes
* Upgrade Dashboard version to fix an issue where the node status was not displayed correctly. [#8771](https://github.com/emqx/emqx/pull/8771)
# 5.0.5
## Bug fixes
* Allow changing the license type from key to file (and vice-versa). [#8598](https://github.com/emqx/emqx/pull/8598)
* Add back http connector config keys `max_retries` `retry_interval` as deprecated fields [#8672](https://github.com/emqx/emqx/issues/8672)
This caused upgrade failure in 5.0.4, because it would fail to boot on configs created from older version.
## Enhancements
* Add `bootstrap_users_file` configuration to add default Dashboard username list, which is only added when EMQX is first started.
* The license is now copied to all nodes in the cluster when it's reloaded. [#8598](https://github.com/emqx/emqx/pull/8598)
* Added a HTTP API to manage licenses. [#8610](https://github.com/emqx/emqx/pull/8610)
* Updated `/nodes` API node_status from `Running/Stopped` to `running/stopped`. [#8642](https://github.com/emqx/emqx/pull/8642)
* Improve handling of placeholder interpolation errors [#8635](https://github.com/emqx/emqx/pull/8635)
* Better logging on unknown object IDs. [#8670](https://github.com/emqx/emqx/pull/8670)
* The bind option support `:1883` style. [#8758](https://github.com/emqx/emqx/pull/8758)
# 5.0.4
## Bug fixes
* The `data/configs/cluster-override.conf` is cleared to 0KB if `hocon_pp:do/2` failed [commits/71f64251](https://github.com/emqx/emqx/pull/8443/commits/71f642518a683cc91a32fd542aafaac6ef915720)
* Improve the health_check for webhooks.
Prior to this change, the webhook only checks the connectivity of the TCP port using `gen_tcp:connect/2`, so
if it's a HTTPs server, we didn't check if TLS handshake was successful.
[commits/6b45d2ea](https://github.com/emqx/emqx/commit/6b45d2ea9fde6d3b4a5b007f7a8c5a1c573d141e)
* The `created_at` field of rules is missing after emqx restarts. [commits/5fc09e6b](https://github.com/emqx/emqx/commit/5fc09e6b950c340243d7be627a0ce1700691221c)
* The rule engine's jq function now works even when the path to the EMQX install dir contains spaces [jq#35](https://github.com/emqx/jq/pull/35) [#8455](https://github.com/emqx/emqx/pull/8455)
* Avoid applying any ACL checks on superusers [#8452](https://github.com/emqx/emqx/pull/8452)
* Fix statistics related system topic name error
* Fix AuthN JWKS SSL schema. Using schema in `emqx_schema`. [#8458](https://github.com/emqx/emqx/pull/8458)
* `sentinel` field should be required when AuthN/AuthZ Redis using sentinel mode. [#8458](https://github.com/emqx/emqx/pull/8458)
* Fix bad swagger format. [#8517](https://github.com/emqx/emqx/pull/8517)
* Fix `chars_limit` is not working when `formatter` is `json`. [#8518](http://github.com/emqx/emqx/pull/8518)
* Ensuring that exhook dispatches the client events are sequential. [#8530](https://github.com/emqx/emqx/pull/8530)
* Avoid using RocksDB backend for persistent sessions when such backend is unavailable. [#8528](https://github.com/emqx/emqx/pull/8528)
* Fix AuthN `cert_subject` and `cert_common_name` placeholder rendering failure. [#8531](https://github.com/emqx/emqx/pull/8531)
* Support listen on an IPv6 address, e.g: [::1]:1883 or ::1:1883. [#8547](https://github.com/emqx/emqx/pull/8547)
* GET '/rules' support for pagination and fuzzy search. [#8472](https://github.com/emqx/emqx/pull/8472)
**‼️ Note** : The previous API only returns array: `[RuleObj1,RuleObj2]`, after updating, it will become
`{"data": [RuleObj1,RuleObj2], "meta":{"count":2, "limit":100, "page":1}`,
which will carry the paging meta information.
* Fix the issue that webhook leaks TCP connections. [ehttpc#34](https://github.com/emqx/ehttpc/pull/34), [#8580](https://github.com/emqx/emqx/pull/8580)
## Enhancements
* Improve the dashboard listener startup log, the listener name is no longer spliced with port information,
and the colon(:) is no longer displayed when IP is not specified. [#8480](https://github.com/emqx/emqx/pull/8480)
* Remove `/configs/listeners` API, use `/listeners/` instead. [#8485](https://github.com/emqx/emqx/pull/8485)
* Optimize performance of builtin database operations in processes with long message queue [#8439](https://github.com/emqx/emqx/pull/8439)
* Improve authentication tracing. [#8554](https://github.com/emqx/emqx/pull/8554)
* Standardize the '/listeners' and `/gateway/<name>/listeners` API fields.
It will introduce some incompatible updates, see [#8571](https://github.com/emqx/emqx/pull/8571)
* Add option to perform GC on connection process after TLS/SSL handshake is performed. [#8637](https://github.com/emqx/emqx/pull/8637)
# 5.0.3
## Bug fixes
* Websocket listener failed to read headers `X-Forwarded-For` and `X-Forwarded-Port` [#8415](https://github.com/emqx/emqx/pull/8415)
* Deleted `cluster_singleton` from MQTT bridge config document. This config is no longer applicable in 5.0 [#8407](https://github.com/emqx/emqx/pull/8407)
* Fix `emqx/emqx:latest` docker image publish to use the Erlang flavor, but not Elixir flavor [#8414](https://github.com/emqx/emqx/pull/8414)
* Changed the `exp` field in JWT auth to be optional rather than required to fix backwards compatability with 4.X releases. [#8425](https://github.com/emqx/emqx/pull/8425)
## Enhancements
* Improve the speed of dashboard's HTTP API routing rule generation, which sometimes causes timeout [#8438](https://github.com/emqx/emqx/pull/8438)
# 5.0.2
Announcement: EMQX team has decided to stop supporting relup for opensource edition.
Going forward, it will be an enterprise-only feature.
Main reason: relup requires carefully crafted upgrade instructions from ALL previous versions.
For example, 4.3 is now at 4.3.16, we have `4.3.0->4.3.16`, `4.3.1->4.3.16`, ... 16 such upgrade paths in total to maintain.
This had been the biggest obstacle for EMQX team to act agile enough in delivering enhancements and fixes.
## Enhancements
## Bug fixes
* Fixed a typo in `bin/emqx` which affects MacOs release when trying to enable Erlang distribution over TLS [#8398](https://github.com/emqx/emqx/pull/8398)
* Restricted shell was accidentally disabled in 5.0.1, it has been added back. [#8396](https://github.com/emqx/emqx/pull/8396)
# 5.0.1
5.0.1 is built on [Erlang/OTP 24.2.1-1](https://github.com/emqx/otp/tree/OTP-24.2.1-1). Same as 5.0.0.
5.0.0 (like 4.4.x) had Erlang/OTP version number in the package name.
This is because we wanted to release different flavor packages (on different Elixir/Erlang/OTP platforms).
However the long package names also causes confusion, as users may not know which to choose if there were more than
one presented at the same time.
Going forward, (starting from 5.0.1), packages will be released in both default (short) and flavored (long) package names.
For example: `emqx-5.0.1-otp24.2.1-1-ubuntu20.04-amd64.tar.gz`,
but only the default one is presented to the users: `emqx-5.0.1-ubuntu20.04-amd64.tar.gz`.
In case anyone wants to try a different flavor package, it can be downlowded from the public s3 bucket,
for example:
https://s3.us-west-2.amazonaws.com/packages.emqx/emqx-ce/v5.0.1/emqx-5.0.1-otp24.2.1-1-ubuntu20.04-arm64.tar.gz
Exceptions:
* Windows package is always presented with short name (currently on Erlang/OTP 24.2.1).
* Elixir package name is flavored with both Elixir and Erlang/OTP version numbers,
for example: `emqx-5.0.1-elixir1.13.4-otp24.2.1-1-ubuntu20.04-amd64.tar.gz`
## Enhancements
* Removed management API auth for prometheus scraping endpoint /api/v5/prometheus/stats [#8299](https://github.com/emqx/emqx/pull/8299)
* Added more TCP options for exhook (gRPC) connections. [#8317](https://github.com/emqx/emqx/pull/8317)
* HTTP Servers used for authentication and authorization will now indicate the result via the response body. [#8374](https://github.com/emqx/emqx/pull/8374) [#8377](https://github.com/emqx/emqx/pull/8377)
* Bulk subscribe/unsubscribe APIs [#8356](https://github.com/emqx/emqx/pull/8356)
* Added exclusive subscription [#8315](https://github.com/emqx/emqx/pull/8315)
* Provide authentication counter metrics [#8352](https://github.com/emqx/emqx/pull/8352) [#8375](https://github.com/emqx/emqx/pull/8375)
* Do not allow admin user self-deletion [#8286](https://github.com/emqx/emqx/pull/8286)
* After restart, ensure to copy `cluster-override.conf` from the clustered node which has the greatest `tnxid`. [#8333](https://github.com/emqx/emqx/pull/8333)
## Bug fixes
* A bug fix ported from 4.x: allow deleting subscriptions from `client.subscribe` hookpoint callback result. [#8304](https://github.com/emqx/emqx/pull/8304) [#8347](https://github.com/emqx/emqx/pull/8377)
* Fixed Erlang distribution over TLS [#8309](https://github.com/emqx/emqx/pull/8309)
* Made possible to override authentication configs from environment variables [#8323](https://github.com/emqx/emqx/pull/8309)
* Made authentication passwords in Mnesia database backward compatible to 4.x, so we can support data migration better. [#8351](https://github.com/emqx/emqx/pull/8351)
* Fix plugins upload for rpm/deb installations [#8379](https://github.com/emqx/emqx/pull/8379)
* Sync data/authz/acl.conf and data/certs from clustered nodes after a new node joins the cluster [#8369](https://github.com/emqx/emqx/pull/8369)
* Ensure auto-retry of failed resources [#8371](https://github.com/emqx/emqx/pull/8371)
* Fix the issue that the count of `packets.connack.auth_error` is inaccurate when the client uses a protocol version below MQTT v5.0 to access [#8178](https://github.com/emqx/emqx/pull/8178)
## Others
* Rate limiter interface is hidden so far, it's subject to a UX redesign.
* QUIC library upgraded to 0.0.14.
* Now the default packages will be released withot otp version number in the package name.
* Renamed config exmpale file name in `etc` dir.

2
Makefile
View File

@ -6,7 +6,7 @@ export EMQX_DEFAULT_BUILDER = ghcr.io/emqx/emqx-builder/5.0-17:1.13.4-24.2.1-1-d
export EMQX_DEFAULT_RUNNER = debian:11-slim export EMQX_DEFAULT_RUNNER = debian:11-slim
export OTP_VSN ?= $(shell $(CURDIR)/scripts/get-otp-vsn.sh) export OTP_VSN ?= $(shell $(CURDIR)/scripts/get-otp-vsn.sh)
export ELIXIR_VSN ?= $(shell $(CURDIR)/scripts/get-elixir-vsn.sh) export ELIXIR_VSN ?= $(shell $(CURDIR)/scripts/get-elixir-vsn.sh)
export EMQX_DASHBOARD_VERSION ?= v1.0.9 export EMQX_DASHBOARD_VERSION ?= v1.1.0
export EMQX_EE_DASHBOARD_VERSION ?= e1.0.0 export EMQX_EE_DASHBOARD_VERSION ?= e1.0.0
export EMQX_REL_FORM ?= tgz export EMQX_REL_FORM ?= tgz
export QUICER_DOWNLOAD_FROM_RELEASE = 1 export QUICER_DOWNLOAD_FROM_RELEASE = 1

12
apps/emqx/i18n/emqx_schema_i18n.conf
View File

@ -882,23 +882,23 @@ When set to true, invalid utf8 strings in for example client ID, topic name, etc
mqtt_max_awaiting_rel { mqtt_max_awaiting_rel {
desc { desc {
en: """Maximum QoS 2 packets (Client -> Broker) awaiting PUBREL.""" en: """For each publisher session, the maximum number of outstanding QoS 2 messages pending on the client to send PUBREL. After reaching this limit, new QoS 2 PUBLISH requests will be rejected with `147(0x93)` until either PUBREL is received or timed out."""
zh: """PUBREL (Client -> Broker) 最大等待队列长度。""" zh: """每个发布者的会话中，都存在一个队列来处理客户端发送的 QoS 2 消息。该队列会存储 QoS 2 消息的报文 ID 直到收到客户端的 PUBREL 或超时，达到队列长度的限制后，新的 QoS 2 消息发布会被拒绝，并返回 `147(0x93)` 错误。"""
} }
label: { label: {
en: """Max Awaiting PUBREL""" en: """Max Awaiting PUBREL"""
zh: """Max Awaiting PUBREL""" zh: """PUBREL 等待队列长度"""
} }
} }
mqtt_await_rel_timeout { mqtt_await_rel_timeout {
desc { desc {
en: """The QoS 2 messages (Client -> Broker) will be dropped if awaiting PUBREL timeout.""" en: """For client to broker QoS 2 message, the time limit for the broker to wait before the `PUBREL` message is received. The wait is aborted after timed out, meaning the packet ID is freed for new `PUBLISH` requests. Receiving a stale `PUBREL` causes a warning level log. Note, the message is delivered to subscribers before entering the wait for PUBREL."""
zh: """PUBREL (Client -> Broker) 最大等待时间，超时则会被丢弃。""" zh: """客户端发布 QoS 2 消息时，服务器等待 `PUBREL` 的最长时延。超过该时长后服务器会放弃等待，该PACKET ID 会被释放，从而允许后续新的 PUBLISH 消息使用。如果超时后收到 PUBREL，服务器将会产生一条告警日志。注意，向订阅客户端转发消息的动作发生在进入等待之前。"""
} }
label: { label: {
en: """Max Awaiting PUBREL TIMEOUT""" en: """Max Awaiting PUBREL TIMEOUT"""
zh: """Max Awaiting PUBREL TIMEOUT""" zh: """PUBREL 最大等待时间"""
} }
} }

2
apps/emqx/include/emqx_release.hrl
View File

@ -32,7 +32,7 @@
%% `apps/emqx/src/bpapi/README.md' %% `apps/emqx/src/bpapi/README.md'
%% Community edition %% Community edition
-define(EMQX_RELEASE_CE, "5.0.8"). -define(EMQX_RELEASE_CE, "5.0.9").
%% Enterprise edition %% Enterprise edition
-define(EMQX_RELEASE_EE, "5.0.0-alpha.1"). -define(EMQX_RELEASE_EE, "5.0.0-alpha.1").

15
apps/emqx/src/emqx_message.erl
View File

@ -74,7 +74,8 @@
to_map/1, to_map/1,
to_log_map/1, to_log_map/1,
to_list/1, to_list/1,
from_map/1 from_map/1,
estimate_size/1
]). ]).
-export_type([message_map/0]). -export_type([message_map/0]).
@ -175,6 +176,18 @@ make(MsgId, From, QoS, Topic, Payload, Flags, Headers) when
timestamp = Now timestamp = Now
}. }.
%% optimistic esitmation of a message size after serialization
%% not including MQTT v5 message headers/user properties etc.
-spec estimate_size(emqx_types:message()) -> non_neg_integer().
estimate_size(#message{topic = Topic, payload = Payload}) ->
FixedHeaderSize = 1,
VarLenSize = 4,
TopicSize = iolist_size(Topic),
PayloadSize = iolist_size(Payload),
PacketIdSize = 2,
TopicLengthSize = 2,
FixedHeaderSize + VarLenSize + TopicLengthSize + TopicSize + PacketIdSize + PayloadSize.
-spec id(emqx_types:message()) -> maybe(binary()). -spec id(emqx_types:message()) -> maybe(binary()).
id(#message{id = Id}) -> Id. id(#message{id = Id}) -> Id.

7
apps/emqx/src/emqx_session.erl
View File

@ -814,12 +814,13 @@ run_terminate_hooks(ClientInfo, Reason, Session) ->
run_hook('session.terminated', [ClientInfo, Reason, info(Session)]). run_hook('session.terminated', [ClientInfo, Reason, info(Session)]).
maybe_redispatch_shared_messages(takenover, _Session) -> maybe_redispatch_shared_messages(takenover, _Session) ->
?tp(debug, ignore_redispatch_shared_messages, #{reason => takenover}),
ok; ok;
maybe_redispatch_shared_messages(kicked, _Session) -> maybe_redispatch_shared_messages(kicked, _Session) ->
?tp(debug, ignore_redispatch_shared_messages, #{reason => kicked}),
ok; ok;
maybe_redispatch_shared_messages(_Reason, #session{inflight = Inflight, mqueue = Q}) -> maybe_redispatch_shared_messages(_Reason, Session) ->
redispatch_shared_messages(Session).
redispatch_shared_messages(#session{inflight = Inflight, mqueue = Q}) ->
AllInflights = emqx_inflight:to_list(fun sort_fun/2, Inflight), AllInflights = emqx_inflight:to_list(fun sort_fun/2, Inflight),
F = fun F = fun
({_PacketId, #inflight_data{message = #message{qos = ?QOS_1} = Msg}}) -> ({_PacketId, #inflight_data{message = #message{qos = ?QOS_1} = Msg}}) ->

137
apps/emqx/test/emqx_shared_sub_SUITE.erl
View File

@ -466,7 +466,7 @@ last_message(ExpectedPayload, Pids) ->
last_message(ExpectedPayload, Pids, Timeout) -> last_message(ExpectedPayload, Pids, Timeout) ->
receive receive
{publish, #{client_pid := Pid, payload := ExpectedPayload}} -> {publish, #{client_pid := Pid, payload := ExpectedPayload}} ->
ct:pal("~p ====== ~p", [Pids, Pid]), ?assert(lists:member(Pid, Pids)),
{true, Pid} {true, Pid}
after Timeout -> after Timeout ->
ct:pal("not yet"), ct:pal("not yet"),
@ -751,11 +751,16 @@ t_dispatch_qos2(Config) when is_list(Config) ->
?assertMatch([{_, _, {ok, 1}}], emqx:publish(Message3)), ?assertMatch([{_, _, {ok, 1}}], emqx:publish(Message3)),
?assertMatch([{_, _, {ok, 1}}], emqx:publish(Message4)), ?assertMatch([{_, _, {ok, 1}}], emqx:publish(Message4)),
%% assert client 2 receives two messages, they are eiter 1,3 or 2,4 depending
%% on if it's picked as the first one for round_robin
MsgRec1 = ?WAIT(2000, {publish, #{client_pid := ConnPid2, payload := P1}}, P1), MsgRec1 = ?WAIT(2000, {publish, #{client_pid := ConnPid2, payload := P1}}, P1),
MsgRec2 = ?WAIT(2000, {publish, #{client_pid := ConnPid2, payload := P2}}, P2), MsgRec2 = ?WAIT(2000, {publish, #{client_pid := ConnPid2, payload := P2}}, P2),
%% assert hello2 > hello1 or hello4 > hello3 case MsgRec2 of
?assert(MsgRec2 > MsgRec1), <<"hello3">> ->
?assertEqual(<<"hello1">>, MsgRec1);
<<"hello4">> ->
?assertEqual(<<"hello2">>, MsgRec1)
end,
sys:resume(ConnPid1), sys:resume(ConnPid1),
%% emqtt subscriber automatically sends PUBREC, but since auto_ack is set to false %% emqtt subscriber automatically sends PUBREC, but since auto_ack is set to false
%% so it will never send PUBCOMP, hence EMQX should not attempt to send %% so it will never send PUBCOMP, hence EMQX should not attempt to send
@ -768,8 +773,14 @@ t_dispatch_qos2(Config) when is_list(Config) ->
kill_process(ConnPid1), kill_process(ConnPid1),
%% client 2 should receive the message %% client 2 should receive the message
MsgRec4 = ?WAIT(2000, {publish, #{client_pid := ConnPid2, payload := P4}}, P4), MsgRec4 = ?WAIT(2000, {publish, #{client_pid := ConnPid2, payload := P4}}, P4),
%% assert hello2 > hello1 or hello4 > hello3 case MsgRec2 of
?assert(MsgRec4 > MsgRec3), <<"hello3">> ->
?assertEqual(<<"hello2">>, MsgRec3),
?assertEqual(<<"hello4">>, MsgRec4);
<<"hello4">> ->
?assertEqual(<<"hello1">>, MsgRec3),
?assertEqual(<<"hello3">>, MsgRec4)
end,
emqtt:stop(ConnPid2), emqtt:stop(ConnPid2),
ok. ok.
@ -818,17 +829,129 @@ t_dispatch_qos0(Config) when is_list(Config) ->
emqtt:stop(ConnPid2), emqtt:stop(ConnPid2),
ok. ok.
t_session_takeover({init, Config}) when is_list(Config) ->
Config;
t_session_takeover({'end', Config}) when is_list(Config) ->
ok;
t_session_takeover(Config) when is_list(Config) ->
Topic = <<"t1/a">>,
ClientId = iolist_to_binary("c" ++ integer_to_list(erlang:system_time())),
Opts = [
{clientid, ClientId},
{auto_ack, true},
{proto_ver, v5},
{clean_start, false},
{properties, #{'Session-Expiry-Interval' => 60}}
],
{ok, ConnPid1} = emqtt:start_link(Opts),
%% with the same client ID, start another client
{ok, ConnPid2} = emqtt:start_link(Opts),
{ok, _} = emqtt:connect(ConnPid1),
emqtt:subscribe(ConnPid1, {<<"$share/t1/", Topic/binary>>, _QoS = 1}),
Message1 = emqx_message:make(<<"dummypub">>, 2, Topic, <<"hello1">>),
Message2 = emqx_message:make(<<"dummypub">>, 2, Topic, <<"hello2">>),
Message3 = emqx_message:make(<<"dummypub">>, 2, Topic, <<"hello3">>),
Message4 = emqx_message:make(<<"dummypub">>, 2, Topic, <<"hello4">>),
%% Make sure client1 is functioning
?assertMatch([_], emqx:publish(Message1)),
{true, _} = last_message(<<"hello1">>, [ConnPid1]),
%% Kill client1
emqtt:stop(ConnPid1),
%% publish another message (should end up in client1's session)
?assertMatch([_], emqx:publish(Message2)),
%% connect client2 (with the same clientid)
%% should trigger session take over
{ok, _} = emqtt:connect(ConnPid2),
?assertMatch([_], emqx:publish(Message3)),
?assertMatch([_], emqx:publish(Message4)),
{true, _} = last_message(<<"hello2">>, [ConnPid2]),
{true, _} = last_message(<<"hello3">>, [ConnPid2]),
{true, _} = last_message(<<"hello4">>, [ConnPid2]),
?assertEqual([], collect_msgs(timer:seconds(2))),
emqtt:stop(ConnPid2),
ok.
t_session_kicked({init, Config}) when is_list(Config) ->
emqx_config:put_zone_conf(default, [mqtt, max_inflight], 1),
Config;
t_session_kicked({'end', Config}) when is_list(Config) ->
emqx_config:put_zone_conf(default, [mqtt, max_inflight], 0);
t_session_kicked(Config) when is_list(Config) ->
ok = ensure_config(round_robin, _AckEnabled = false),
Topic = <<"foo/bar/1">>,
ClientId1 = <<"ClientId1">>,
ClientId2 = <<"ClientId2">>,
{ok, ConnPid1} = emqtt:start_link([{clientid, ClientId1}, {auto_ack, false}]),
{ok, ConnPid2} = emqtt:start_link([{clientid, ClientId2}, {auto_ack, true}]),
{ok, _} = emqtt:connect(ConnPid1),
{ok, _} = emqtt:connect(ConnPid2),
emqtt:subscribe(ConnPid1, {<<"$share/group/foo/bar/#">>, 2}),
emqtt:subscribe(ConnPid2, {<<"$share/group/foo/bar/#">>, 2}),
Message1 = emqx_message:make(ClientId1, 2, Topic, <<"hello1">>),
Message2 = emqx_message:make(ClientId1, 2, Topic, <<"hello2">>),
Message3 = emqx_message:make(ClientId1, 2, Topic, <<"hello3">>),
Message4 = emqx_message:make(ClientId1, 2, Topic, <<"hello4">>),
ct:sleep(100),
ok = sys:suspend(ConnPid1),
%% One message is inflight
?assertMatch([{_, _, {ok, 1}}], emqx:publish(Message1)),
?assertMatch([{_, _, {ok, 1}}], emqx:publish(Message2)),
?assertMatch([{_, _, {ok, 1}}], emqx:publish(Message3)),
?assertMatch([{_, _, {ok, 1}}], emqx:publish(Message4)),
%% assert client 2 receives two messages, they are eiter 1,3 or 2,4 depending
%% on if it's picked as the first one for round_robin
MsgRec1 = ?WAIT(2000, {publish, #{client_pid := ConnPid2, payload := P1}}, P1),
MsgRec2 = ?WAIT(2000, {publish, #{client_pid := ConnPid2, payload := P2}}, P2),
case MsgRec2 of
<<"hello3">> ->
?assertEqual(<<"hello1">>, MsgRec1);
<<"hello4">> ->
?assertEqual(<<"hello2">>, MsgRec1)
end,
sys:resume(ConnPid1),
%% emqtt subscriber automatically sends PUBREC, but since auto_ack is set to false
%% so it will never send PUBCOMP, hence EMQX should not attempt to send
%% the 4th message yet since max_inflight is 1.
MsgRec3 = ?WAIT(2000, {publish, #{client_pid := ConnPid1, payload := P3}}, P3),
case MsgRec2 of
<<"hello3">> ->
?assertEqual(<<"hello2">>, MsgRec3);
<<"hello4">> ->
?assertEqual(<<"hello1">>, MsgRec3)
end,
%% no message expected
?assertEqual([], collect_msgs(0)),
%% now kick client 1
kill_process(ConnPid1, fun(_Pid) -> emqx_cm:kick_session(ClientId1) end),
%% client 2 should NOT receive the message
?assertEqual([], collect_msgs(1000)),
emqtt:stop(ConnPid2),
?assertEqual([], collect_msgs(0)),
ok.
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% help functions %% help functions
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
kill_process(Pid) -> kill_process(Pid) ->
kill_process(Pid, fun(_) -> erlang:exit(Pid, kill) end).
kill_process(Pid, WithFun) ->
_ = unlink(Pid), _ = unlink(Pid),
_ = monitor(process, Pid), _ = monitor(process, Pid),
erlang:exit(Pid, kill), _ = WithFun(Pid),
receive receive
{'DOWN', _, process, Pid, _} -> {'DOWN', _, process, Pid, _} ->
ok ok
after 10_000 ->
error(timeout)
end. end.
collect_msgs(Timeout) -> collect_msgs(Timeout) ->

127
apps/emqx_management/i18n/emqx_mgmt_api_publish_i18n.conf Normal file
View File

@ -0,0 +1,127 @@
emqx_mgmt_api_publish {
publish_api {
desc {
en: """
Publish one message.<br/>
Possible HTTP status response codes are:<br/>
<code>200</code>: The message is delivered to at least one subscriber;<br/>
<code>202</code>: No matched subscribers;<br/>
<code>400</code>: Message is invalid. for example bad topic name, or QoS is out of range;<br/>
<code>503</code>: Failed to deliver the message to subscriber(s);<br/>
"""
zh: """
发布一个消息。<br/>
可能的 HTTP 状态码如下：<br/>
200: 消息被成功发送到至少一个订阅。<br/>
202: 没有匹配到任何订阅。<br/>
400: 消息编码错误，如非法主题，或 QoS 超出范围等。<br/>
503: 服务重启等过程中导致转发失败。<br/><br/>
"""
}
}
publish_bulk_api {
desc {
en: """
Publish a batch of messages.<br/>
Possible HTTP response status code are:<br/>
200: All messages are delivered to at least one subscriber;<br/>
202: At least one message was not delivered to any subscriber;<br/>
400: At least one message is invalid. For example bad topic name, or QoS is out of range;<br/>
503: Failed to deliver at least one of the messages;<br/>
In case there is at lest one invalid message in the batch, the HTTP response body
is the same as for <code>/publish</code> API.<br/>
Otherwise the HTTP response body is an array of JSON objects indicating the publish
result of each individual message in the batch.
"""
zh: """
批量发布一组消息。<br/>
可能的 HTTP 状态码如下：<br/>
200: 所有的消息都被成功发送到至少一个订阅。<br/>
202: 至少有一个消息没有匹配到任何订阅。<br/>
400: 至少有一个消息编码错误，如非法主题，或 QoS 超出范围等。<br/>
503: 至少有一个小因为服务重启的原因导致转发失败。<br/>
请求的 Body 或者 Body 中包含的某个消息无法通过 API 规范的类型检查时，HTTP 响应的消息与发布单个消息的 API
<code>/publish</code> 是一样的。
如果所有的消息都是合法的，那么 HTTP 返回的内容是一个 JSON 数组，每个元素代表了该消息转发的状态。
"""
}
}
topic_name {
desc {
en: "Topic Name"
zh: "主题名称"
}
}
qos {
desc {
en: "MQTT message QoS"
zh: "MQTT 消息的 QoS"
}
}
clientid {
desc {
en: "Each message can be published as if it is done on behalf of an MQTT client whos ID can be specified in this field."
zh: "每个消息都可以带上一个 MQTT 客户端 ID，用于模拟 MQTT 客户端的发布行为。"
}
}
payload {
desc {
en: "The MQTT message payload."
zh: "MQTT 消息体。"
}
}
retain {
desc {
en: "A boolean field to indicate if this message should be retained."
zh: "布尔型字段，用于表示该消息是否保留消息。"
}
}
payload_encoding {
desc {
en: "MQTT Payload Encoding, <code>base64</code> or <code>plain</code>. When set to <code>base64</code>, the message is decoded before it is published."
zh: "MQTT 消息体的编码方式，可以是 <code>base64</code> 或 <code>plain</code>。当设置为 <code>base64</code> 时，消息在发布前会先被解码。"
}
}
message_id {
desc {
en: "A globally unique message ID for correlation/tracing."
zh: "全局唯一的一个消息 ID，方便用于关联和追踪。"
}
}
reason_code {
desc {
en: """
The MQTT reason code, as the same ones used in PUBACK packet.<br/>
Currently supported codes are:<br/>
16(0x10): No matching subscribers;<br/>
131(0x81): Error happened when dispatching the message. e.g. during EMQX restart;<br/>
144(0x90): Topic name invalid;<br/>
151(0x97): Publish rate limited, or message size exceeded limit. The global size limit can be configured with <code>mqtt.max_packet_size</code><br/>
NOTE: The message size is estimated with the received topic and payload size, meaning the actual size of serialized bytes (when sent to MQTT subscriber)
might be slightly over the limit.
"""
zh: """
MQTT 消息发布的错误码，这些错误码也是 MQTT 规范中 PUBACK 消息可能携带的错误码。<br/>
当前支持如下错误码：<br/>
16(0x10)：没能匹配到任何订阅；<br/>
131(0x81)：消息转发时发生错误，例如 EMQX 服务重启；<br/>
144(0x90)：主题名称非法；<br/>
151(0x97)：受到了速率限制，或者消息尺寸过大。全局消息大小限制可以通过配置项 <code>mqtt.max_packet_size</code> 来进行修改。<br/>
注意：消息尺寸的是通过主题和消息体的字节数进行估算的。具体发布时所占用的字节数可能会稍大于这个估算的值。
"""
}
}
error_message {
desc {
en: "Describes the failure reason in detail."
zh: "失败的详细原因。"
}
}
}

241
apps/emqx_management/src/emqx_mgmt_api_publish.erl
View File

@ -16,7 +16,15 @@
-module(emqx_mgmt_api_publish). -module(emqx_mgmt_api_publish).
-include_lib("emqx/include/emqx.hrl"). -include_lib("emqx/include/emqx.hrl").
-include_lib("emqx/include/emqx_mqtt.hrl").
-include_lib("typerefl/include/types.hrl"). -include_lib("typerefl/include/types.hrl").
-include_lib("emqx/include/logger.hrl").
-include_lib("hocon/include/hoconsc.hrl").
-define(ALL_IS_WELL, 200).
-define(PARTIALLY_OK, 202).
-define(BAD_REQUEST, 400).
-define(DISPATCH_ERROR, 503).
-behaviour(minirest_api). -behaviour(minirest_api).
@ -42,11 +50,14 @@ schema("/publish") ->
#{ #{
'operationId' => publish, 'operationId' => publish,
post => #{ post => #{
description => <<"Publish Message">>, description => ?DESC(publish_api),
tags => [<<"Publish">>], tags => [<<"Publish">>],
'requestBody' => hoconsc:mk(hoconsc:ref(?MODULE, publish_message)), 'requestBody' => hoconsc:mk(hoconsc:ref(?MODULE, publish_message)),
responses => #{ responses => #{
200 => hoconsc:mk(hoconsc:ref(?MODULE, publish_message_info)) ?ALL_IS_WELL => hoconsc:mk(hoconsc:ref(?MODULE, publish_ok)),
?PARTIALLY_OK => hoconsc:mk(hoconsc:ref(?MODULE, publish_error)),
?BAD_REQUEST => bad_request_schema(),
?DISPATCH_ERROR => hoconsc:mk(hoconsc:ref(?MODULE, publish_error))
} }
} }
}; };
@ -54,44 +65,58 @@ schema("/publish/bulk") ->
#{ #{
'operationId' => publish_batch, 'operationId' => publish_batch,
post => #{ post => #{
description => <<"Publish Messages">>, description => ?DESC(publish_bulk_api),
tags => [<<"Publish">>], tags => [<<"Publish">>],
'requestBody' => hoconsc:mk(hoconsc:array(hoconsc:ref(?MODULE, publish_message)), #{}), 'requestBody' => hoconsc:mk(hoconsc:array(hoconsc:ref(?MODULE, publish_message)), #{}),
responses => #{ responses => #{
200 => hoconsc:mk(hoconsc:array(hoconsc:ref(?MODULE, publish_message_info)), #{}) ?ALL_IS_WELL => hoconsc:mk(hoconsc:array(hoconsc:ref(?MODULE, publish_ok)), #{}),
?PARTIALLY_OK => hoconsc:mk(
hoconsc:array(hoconsc:ref(?MODULE, publish_error)), #{}
),
?BAD_REQUEST => bad_request_schema(),
?DISPATCH_ERROR => hoconsc:mk(
hoconsc:array(hoconsc:ref(?MODULE, publish_error)), #{}
)
} }
} }
}. }.
bad_request_schema() ->
Union = hoconsc:union([
hoconsc:ref(?MODULE, bad_request),
hoconsc:array(hoconsc:ref(?MODULE, publish_error))
]),
hoconsc:mk(Union, #{}).
fields(message) -> fields(message) ->
[ [
{topic, {topic,
hoconsc:mk(binary(), #{ hoconsc:mk(binary(), #{
desc => <<"Topic Name">>, desc => ?DESC(topic_name),
required => true, required => true,
example => <<"api/example/topic">> example => <<"api/example/topic">>
})}, })},
{qos, {qos,
hoconsc:mk(emqx_schema:qos(), #{ hoconsc:mk(emqx_schema:qos(), #{
desc => <<"MQTT QoS">>, desc => ?DESC(qos),
required => false, required => false,
default => 0 default => 0
})}, })},
{clientid, {clientid,
hoconsc:mk(binary(), #{ hoconsc:mk(binary(), #{
desc => <<"From client ID">>, desc => ?DESC(clientid),
required => false, required => false,
example => <<"api_example_client">> example => <<"api_example_client">>
})}, })},
{payload, {payload,
hoconsc:mk(binary(), #{ hoconsc:mk(binary(), #{
desc => <<"MQTT Payload">>, desc => ?DESC(payload),
required => true, required => true,
example => <<"hello emqx api">> example => <<"hello emqx api">>
})}, })},
{retain, {retain,
hoconsc:mk(boolean(), #{ hoconsc:mk(boolean(), #{
desc => <<"MQTT Retain Message">>, desc => ?DESC(retain),
required => false, required => false,
default => false default => false
})} })}
@ -100,53 +125,196 @@ fields(publish_message) ->
[ [
{payload_encoding, {payload_encoding,
hoconsc:mk(hoconsc:enum([plain, base64]), #{ hoconsc:mk(hoconsc:enum([plain, base64]), #{
desc => <<"MQTT Payload Encoding, base64 or plain">>, desc => ?DESC(payload_encoding),
required => false, required => false,
default => plain default => plain
})} })}
] ++ fields(message); ] ++ fields(message);
fields(publish_message_info) -> fields(publish_ok) ->
[ [
{id, {id,
hoconsc:mk(binary(), #{ hoconsc:mk(binary(), #{
desc => <<"A globally unique message ID">> desc => ?DESC(message_id)
})}
];
fields(publish_error) ->
[
{reason_code,
hoconsc:mk(integer(), #{
desc => ?DESC(reason_code),
example => 16
})},
{message,
hoconsc:mk(binary(), #{
desc => ?DESC(error_message),
example => <<"no_matching_subscribers">>
})}
];
fields(bad_request) ->
[
{code,
hoconsc:mk(string(), #{
desc => <<"BAD_REQUEST">>
})},
{message,
hoconsc:mk(binary(), #{
desc => ?DESC(error_message)
})} })}
]. ].
publish(post, #{body := Body}) -> publish(post, #{body := Body}) ->
case message(Body) of case message(Body) of
{ok, Message} -> {ok, Message} ->
_ = emqx_mgmt:publish(Message), Res = emqx_mgmt:publish(Message),
{200, format_message_response(Message)}; publish_result_to_http_reply(Message, Res);
{error, R} -> {error, Reason} ->
{400, 'BAD_REQUEST', to_binary(R)} {?BAD_REQUEST, make_bad_req_reply(Reason)}
end. end.
publish_batch(post, #{body := Body}) -> publish_batch(post, #{body := Body}) ->
case messages(Body) of case messages(Body) of
{ok, Messages} -> {ok, Messages} ->
_ = [emqx_mgmt:publish(Message) || Message <- Messages], ResList = lists:map(
{200, format_message_response(Messages)}; fun(Message) ->
{error, R} -> Res = emqx_mgmt:publish(Message),
{400, 'BAD_REQUEST', to_binary(R)} publish_result_to_http_reply(Message, Res)
end,
Messages
),
publish_results_to_http_reply(ResList);
{error, Reason} ->
{?BAD_REQUEST, make_bad_req_reply(Reason)}
end. end.
make_bad_req_reply(invalid_topic_name) ->
make_publish_error_response(?RC_TOPIC_NAME_INVALID);
make_bad_req_reply(packet_too_large) ->
%% 0x95 RC_PACKET_TOO_LARGE is not a PUBACK reason code
%% This is why we use RC_QUOTA_EXCEEDED instead
make_publish_error_response(?RC_QUOTA_EXCEEDED, packet_too_large);
make_bad_req_reply(Reason) ->
make_publish_error_response(?RC_IMPLEMENTATION_SPECIFIC_ERROR, to_binary(Reason)).
-spec is_ok_deliver({_NodeOrShare, _MatchedTopic, emqx_types:deliver_result()}) -> boolean().
is_ok_deliver({_NodeOrShare, _MatchedTopic, ok}) -> true;
is_ok_deliver({_NodeOrShare, _MatchedTopic, {ok, _}}) -> true;
is_ok_deliver({_NodeOrShare, _MatchedTopic, {error, _}}) -> false.
%% @hidden Map MQTT publish result reason code to HTTP status code.
%% MQTT reason code | Description | HTTP status code
%% 0 Success 200
%% 16 No matching subscribers 202
%% 128 Unspecified error 406
%% 131 Implementation specific error 406
%% 144 Topic Name invalid 400
%% 151 Quota exceeded 400
%%
%% %%%%%% Below error codes are not implemented so far %%%%
%%
%% If HTTP request passes HTTP authentication, it is considered trusted.
%% In the future, we may choose to check ACL for the provided MQTT Client ID
%% 135 Not authorized 401
%%
%% %%%%%% Below error codes are not applicable %%%%%%%
%%
%% No user specified packet ID, so there should be no packet ID error
%% 145 Packet identifier is in use 400
%%
%% No preceding payload format indicator to compare against.
%% Content-Type check should be done at HTTP layer but not here.
%% 153 Payload format invalid 400
publish_result_to_http_reply(_Message, []) ->
%% matched no subscriber
{?PARTIALLY_OK, make_publish_error_response(?RC_NO_MATCHING_SUBSCRIBERS)};
publish_result_to_http_reply(Message, PublishResult) ->
case lists:any(fun is_ok_deliver/1, PublishResult) of
true ->
%% delivered to at least one subscriber
OkBody = make_publish_response(Message),
{?ALL_IS_WELL, OkBody};
false ->
%% this is quite unusual, matched, but failed to deliver
%% if this happens, the publish result log can be helpful
%% to idnetify the reason why publish failed
%% e.g. during emqx restart
ReasonString = <<"failed_to_dispatch">>,
ErrorBody = make_publish_error_response(
?RC_IMPLEMENTATION_SPECIFIC_ERROR, ReasonString
),
?SLOG(warning, #{
msg => ReasonString,
message_id => emqx_message:id(Message),
results => PublishResult
}),
{?DISPATCH_ERROR, ErrorBody}
end.
%% @hidden Reply batch publish result.
%% 200 if all published OK.
%% 202 if at least one message matched no subscribers.
%% 503 for temp errors duing EMQX restart
publish_results_to_http_reply([_ | _] = ResList) ->
{Codes0, BodyL} = lists:unzip(ResList),
Codes = lists:usort(Codes0),
HasFailure = lists:member(?DISPATCH_ERROR, Codes),
All200 = (Codes =:= [?ALL_IS_WELL]),
Code =
case All200 of
true ->
%% All OK
?ALL_IS_WELL;
false when not HasFailure ->
%% Partially OK
?PARTIALLY_OK;
false ->
%% At least one failed
?DISPATCH_ERROR
end,
{Code, BodyL}.
message(Map) -> message(Map) ->
try
make_message(Map)
catch
throw:Reason ->
{error, Reason}
end.
make_message(Map) ->
Encoding = maps:get(<<"payload_encoding">>, Map, plain), Encoding = maps:get(<<"payload_encoding">>, Map, plain),
case encode_payload(Encoding, maps:get(<<"payload">>, Map)) of case decode_payload(Encoding, maps:get(<<"payload">>, Map)) of
{ok, Payload} -> {ok, Payload} ->
From = maps:get(<<"clientid">>, Map, http_api), From = maps:get(<<"clientid">>, Map, http_api),
QoS = maps:get(<<"qos">>, Map, 0), QoS = maps:get(<<"qos">>, Map, 0),
Topic = maps:get(<<"topic">>, Map), Topic = maps:get(<<"topic">>, Map),
Retain = maps:get(<<"retain">>, Map, false), Retain = maps:get(<<"retain">>, Map, false),
{ok, emqx_message:make(From, QoS, Topic, Payload, #{retain => Retain}, #{})}; try
_ = emqx_topic:validate(name, Topic)
catch
error:_Reason ->
throw(invalid_topic_name)
end,
Message = emqx_message:make(From, QoS, Topic, Payload, #{retain => Retain}, #{}),
Size = emqx_message:estimate_size(Message),
(Size > size_limit()) andalso throw(packet_too_large),
{ok, Message};
{error, R} -> {error, R} ->
{error, R} {error, R}
end. end.
encode_payload(plain, Payload) -> %% get the global packet size limit since HTTP API does not belong to any zone.
size_limit() ->
try
emqx_config:get([mqtt, max_packet_size])
catch
_:_ ->
%% leave 1000 bytes for topic name etc.
?MAX_PACKET_SIZE
end.
decode_payload(plain, Payload) ->
{ok, Payload}; {ok, Payload};
encode_payload(base64, Payload) -> decode_payload(base64, Payload) ->
try try
{ok, base64:decode(Payload)} {ok, base64:decode(Payload)}
catch catch
@ -154,6 +322,8 @@ encode_payload(base64, Payload) ->
{error, {decode_base64_payload_failed, Payload}} {error, {decode_base64_payload_failed, Payload}}
end. end.
messages([]) ->
{errror, <<"empty_batch">>};
messages(List) -> messages(List) ->
messages(List, []). messages(List, []).
@ -167,10 +337,23 @@ messages([MessageMap | List], Res) ->
{error, R} {error, R}
end. end.
format_message_response(Messages) when is_list(Messages) -> make_publish_response(#message{id = ID}) ->
[format_message_response(Message) || Message <- Messages]; #{
format_message_response(#message{id = ID}) -> id => emqx_guid:to_hexstr(ID)
#{id => emqx_guid:to_hexstr(ID)}. }.
make_publish_error_response(ReasonCode) ->
make_publish_error_response(ReasonCode, emqx_reason_codes:name(ReasonCode)).
make_publish_error_response(ReasonCode, Msg) ->
#{
reason_code => ReasonCode,
message => to_binary(Msg)
}.
to_binary(Atom) when is_atom(Atom) ->
atom_to_binary(Atom);
to_binary(Msg) when is_binary(Msg) ->
Msg;
to_binary(Term) -> to_binary(Term) ->
list_to_binary(io_lib:format("~p", [Term])). list_to_binary(io_lib:format("~0p", [Term])).

246
apps/emqx_management/test/emqx_mgmt_api_publish_SUITE.erl
View File

@ -19,6 +19,7 @@
-compile(nowarn_export_all). -compile(nowarn_export_all).
-include_lib("eunit/include/eunit.hrl"). -include_lib("eunit/include/eunit.hrl").
-include_lib("emqx/include/emqx_mqtt.hrl").
-define(CLIENTID, <<"api_clientid">>). -define(CLIENTID, <<"api_clientid">>).
-define(USERNAME, <<"api_username">>). -define(USERNAME, <<"api_username">>).
@ -36,6 +37,16 @@ init_per_suite(Config) ->
end_per_suite(_) -> end_per_suite(_) ->
emqx_mgmt_api_test_util:end_suite(). emqx_mgmt_api_test_util:end_suite().
init_per_testcase(Case, Config) ->
?MODULE:Case({init, Config}).
end_per_testcase(Case, Config) ->
?MODULE:Case({'end', Config}).
t_publish_api({init, Config}) ->
Config;
t_publish_api({'end', _Config}) ->
ok;
t_publish_api(_) -> t_publish_api(_) ->
{ok, Client} = emqtt:start_link(#{ {ok, Client} = emqtt:start_link(#{
username => <<"api_username">>, clientid => <<"api_clientid">> username => <<"api_username">>, clientid => <<"api_clientid">>
@ -48,11 +59,113 @@ t_publish_api(_) ->
Auth = emqx_mgmt_api_test_util:auth_header_(), Auth = emqx_mgmt_api_test_util:auth_header_(),
Body = #{topic => ?TOPIC1, payload => Payload}, Body = #{topic => ?TOPIC1, payload => Payload},
{ok, Response} = emqx_mgmt_api_test_util:request_api(post, Path, "", Auth, Body), {ok, Response} = emqx_mgmt_api_test_util:request_api(post, Path, "", Auth, Body),
ResponseMap = emqx_json:decode(Response, [return_maps]), ResponseMap = decode_json(Response),
?assertEqual([<<"id">>], maps:keys(ResponseMap)), ?assertEqual([<<"id">>], lists:sort(maps:keys(ResponseMap))),
?assertEqual(ok, receive_assert(?TOPIC1, 0, Payload)), ?assertEqual(ok, receive_assert(?TOPIC1, 0, Payload)),
emqtt:disconnect(Client). emqtt:stop(Client).
t_publish_no_subscriber({init, Config}) ->
Config;
t_publish_no_subscriber({'end', _Config}) ->
ok;
t_publish_no_subscriber(_) ->
Payload = <<"hello">>,
Path = emqx_mgmt_api_test_util:api_path(["publish"]),
Auth = emqx_mgmt_api_test_util:auth_header_(),
Body = #{topic => ?TOPIC1, payload => Payload},
{ok, Response} = emqx_mgmt_api_test_util:request_api(post, Path, "", Auth, Body),
ResponseMap = decode_json(Response),
?assertEqual([<<"message">>, <<"reason_code">>], lists:sort(maps:keys(ResponseMap))),
?assertMatch(#{<<"reason_code">> := ?RC_NO_MATCHING_SUBSCRIBERS}, ResponseMap),
ok.
t_publish_bad_topic({init, Config}) ->
Config;
t_publish_bad_topic({'end', _Config}) ->
ok;
t_publish_bad_topic(_) ->
Payload = <<"hello">>,
Path = emqx_mgmt_api_test_util:api_path(["publish"]),
Auth = emqx_mgmt_api_test_util:auth_header_(),
Body = #{topic => <<"not/a+/valid/topic">>, payload => Payload},
?assertMatch(
{error, {_, 400, _}}, emqx_mgmt_api_test_util:request_api(post, Path, "", Auth, Body)
).
t_publish_bad_base64({init, Config}) ->
Config;
t_publish_bad_base64({'end', _Config}) ->
ok;
t_publish_bad_base64(_) ->
%% not a valid base64
Payload = <<"hello">>,
Path = emqx_mgmt_api_test_util:api_path(["publish"]),
Auth = emqx_mgmt_api_test_util:auth_header_(),
Body = #{
topic => <<"not/a+/valid/topic">>, payload => Payload, payload_encoding => <<"base64">>
},
?assertMatch(
{error, {_, 400, _}}, emqx_mgmt_api_test_util:request_api(post, Path, "", Auth, Body)
).
t_publish_too_large({init, Config}) ->
MaxPacketSize = 100,
meck:new(emqx_config, [no_link, passthrough, no_history]),
meck:expect(emqx_config, get, fun
([mqtt, max_packet_size]) ->
MaxPacketSize;
(Other) ->
meck:passthrough(Other)
end),
[{max_packet_size, MaxPacketSize} | Config];
t_publish_too_large({'end', _Config}) ->
meck:unload(emqx_config),
ok;
t_publish_too_large(Config) ->
MaxPacketSize = proplists:get_value(max_packet_size, Config),
Payload = lists:duplicate(MaxPacketSize, $0),
Path = emqx_mgmt_api_test_util:api_path(["publish"]),
Auth = emqx_mgmt_api_test_util:auth_header_(),
Body = #{topic => <<"random/topic">>, payload => Payload},
{error, {Summary, _Headers, ResponseBody}} =
emqx_mgmt_api_test_util:request_api(
post,
Path,
"",
Auth,
Body,
#{return_body => true}
),
?assertMatch({_, 400, _}, Summary),
?assertMatch(
#{
<<"reason_code">> := ?RC_QUOTA_EXCEEDED,
<<"message">> := <<"packet_too_large">>
},
decode_json(ResponseBody)
),
ok.
t_publish_bad_topic_bulk({init, Config}) ->
Config;
t_publish_bad_topic_bulk({'end', _Config}) ->
ok;
t_publish_bad_topic_bulk(_Config) ->
Payload = <<"hello">>,
Path = emqx_mgmt_api_test_util:api_path(["publish", "bulk"]),
Auth = emqx_mgmt_api_test_util:auth_header_(),
Body = [
#{topic => <<"not/a+/valid/topic">>, payload => Payload},
#{topic => <<"good/topic">>, payload => Payload}
],
?assertMatch(
{error, {_, 400, _}}, emqx_mgmt_api_test_util:request_api(post, Path, "", Auth, Body)
).
t_publish_bulk_api({init, Config}) ->
Config;
t_publish_bulk_api({'end', _Config}) ->
ok;
t_publish_bulk_api(_) -> t_publish_bulk_api(_) ->
{ok, Client} = emqtt:start_link(#{ {ok, Client} = emqtt:start_link(#{
username => <<"api_username">>, clientid => <<"api_clientid">> username => <<"api_username">>, clientid => <<"api_clientid">>
@ -63,19 +176,135 @@ t_publish_bulk_api(_) ->
Payload = <<"hello">>, Payload = <<"hello">>,
Path = emqx_mgmt_api_test_util:api_path(["publish", "bulk"]), Path = emqx_mgmt_api_test_util:api_path(["publish", "bulk"]),
Auth = emqx_mgmt_api_test_util:auth_header_(), Auth = emqx_mgmt_api_test_util:auth_header_(),
Body = [#{topic => ?TOPIC1, payload => Payload}, #{topic => ?TOPIC2, payload => Payload}], Body = [
#{
topic => ?TOPIC1,
payload => Payload,
payload_encoding => plain
},
#{
topic => ?TOPIC2,
payload => base64:encode(Payload),
payload_encoding => base64
}
],
{ok, Response} = emqx_mgmt_api_test_util:request_api(post, Path, "", Auth, Body), {ok, Response} = emqx_mgmt_api_test_util:request_api(post, Path, "", Auth, Body),
ResponseList = emqx_json:decode(Response, [return_maps]), ResponseList = decode_json(Response),
?assertEqual(2, erlang:length(ResponseList)), ?assertEqual(2, erlang:length(ResponseList)),
lists:foreach( lists:foreach(
fun(ResponseMap) -> fun(ResponseMap) ->
?assertEqual([<<"id">>], maps:keys(ResponseMap)) ?assertMatch(
[<<"id">>], lists:sort(maps:keys(ResponseMap))
)
end, end,
ResponseList ResponseList
), ),
?assertEqual(ok, receive_assert(?TOPIC1, 0, Payload)), ?assertEqual(ok, receive_assert(?TOPIC1, 0, Payload)),
?assertEqual(ok, receive_assert(?TOPIC2, 0, Payload)), ?assertEqual(ok, receive_assert(?TOPIC2, 0, Payload)),
emqtt:disconnect(Client). emqtt:stop(Client).
t_publish_no_subscriber_bulk({init, Config}) ->
Config;
t_publish_no_subscriber_bulk({'end', _Config}) ->
ok;
t_publish_no_subscriber_bulk(_) ->
{ok, Client} = emqtt:start_link(#{
username => <<"api_username">>, clientid => <<"api_clientid">>
}),
{ok, _} = emqtt:connect(Client),
{ok, _, [0]} = emqtt:subscribe(Client, ?TOPIC1),
{ok, _, [0]} = emqtt:subscribe(Client, ?TOPIC2),
Payload = <<"hello">>,
Path = emqx_mgmt_api_test_util:api_path(["publish", "bulk"]),
Auth = emqx_mgmt_api_test_util:auth_header_(),
Body = [
#{topic => ?TOPIC1, payload => Payload},
#{topic => ?TOPIC2, payload => Payload},
#{topic => <<"no/subscrbier/topic">>, payload => Payload}
],
{ok, Response} = emqx_mgmt_api_test_util:request_api(post, Path, "", Auth, Body),
ResponseList = decode_json(Response),
?assertMatch(
[
#{<<"id">> := _},
#{<<"id">> := _},
#{<<"message">> := <<"no_matching_subscribers">>}
],
ResponseList
),
?assertEqual(ok, receive_assert(?TOPIC1, 0, Payload)),
?assertEqual(ok, receive_assert(?TOPIC2, 0, Payload)),
emqtt:stop(Client).
t_publish_bulk_dispatch_one_message_invalid_topic({init, Config}) ->
Config;
t_publish_bulk_dispatch_one_message_invalid_topic({'end', _Config}) ->
ok;
t_publish_bulk_dispatch_one_message_invalid_topic(Config) when is_list(Config) ->
Payload = <<"hello">>,
Path = emqx_mgmt_api_test_util:api_path(["publish", "bulk"]),
Auth = emqx_mgmt_api_test_util:auth_header_(),
Body = [
#{topic => ?TOPIC1, payload => Payload},
#{topic => ?TOPIC2, payload => Payload},
#{topic => <<"bad/#/topic">>, payload => Payload}
],
{error, {Summary, _Headers, ResponseBody}} =
emqx_mgmt_api_test_util:request_api(
post,
Path,
"",
Auth,
Body,
#{return_body => true}
),
?assertMatch({_, 400, _}, Summary),
?assertMatch(
#{<<"reason_code">> := ?RC_TOPIC_NAME_INVALID},
decode_json(ResponseBody)
).
t_publish_bulk_dispatch_failure({init, Config}) ->
meck:new(emqx, [no_link, passthrough, no_history]),
meck:expect(emqx, is_running, fun() -> false end),
Config;
t_publish_bulk_dispatch_failure({'end', _Config}) ->
meck:unload(emqx),
ok;
t_publish_bulk_dispatch_failure(Config) when is_list(Config) ->
{ok, Client} = emqtt:start_link(#{
username => <<"api_username">>, clientid => <<"api_clientid">>
}),
{ok, _} = emqtt:connect(Client),
{ok, _, [0]} = emqtt:subscribe(Client, ?TOPIC1),
{ok, _, [0]} = emqtt:subscribe(Client, ?TOPIC2),
Payload = <<"hello">>,
Path = emqx_mgmt_api_test_util:api_path(["publish", "bulk"]),
Auth = emqx_mgmt_api_test_util:auth_header_(),
Body = [
#{topic => ?TOPIC1, payload => Payload},
#{topic => ?TOPIC2, payload => Payload},
#{topic => <<"no/subscrbier/topic">>, payload => Payload}
],
{error, {Summary, _Headers, ResponseBody}} =
emqx_mgmt_api_test_util:request_api(
post,
Path,
"",
Auth,
Body,
#{return_body => true}
),
?assertMatch({_, 503, _}, Summary),
?assertMatch(
[
#{<<"reason_code">> := ?RC_IMPLEMENTATION_SPECIFIC_ERROR},
#{<<"reason_code">> := ?RC_IMPLEMENTATION_SPECIFIC_ERROR},
#{<<"reason_code">> := ?RC_NO_MATCHING_SUBSCRIBERS}
],
decode_json(ResponseBody)
),
emqtt:stop(Client).
receive_assert(Topic, Qos, Payload) -> receive_assert(Topic, Qos, Payload) ->
receive receive
@ -90,3 +319,6 @@ receive_assert(Topic, Qos, Payload) ->
after 5000 -> after 5000 ->
timeout timeout
end. end.
decode_json(In) ->
emqx_json:decode(In, [return_maps]).

18
apps/emqx_management/test/emqx_mgmt_api_test_util.erl
View File

@ -65,8 +65,11 @@ request_api(Method, Url, QueryParams, AuthOrHeaders, []) when
"" -> Url; "" -> Url;
_ -> Url ++ "?" ++ QueryParams _ -> Url ++ "?" ++ QueryParams
end, end,
do_request_api(Method, {NewUrl, build_http_header(AuthOrHeaders)}); do_request_api(Method, {NewUrl, build_http_header(AuthOrHeaders)}, #{});
request_api(Method, Url, QueryParams, AuthOrHeaders, Body) when request_api(Method, Url, QueryParams, AuthOrHeaders, Body) ->
request_api(Method, Url, QueryParams, AuthOrHeaders, Body, #{}).
request_api(Method, Url, QueryParams, AuthOrHeaders, Body, Opts) when
(Method =:= post) orelse (Method =:= post) orelse
(Method =:= patch) orelse (Method =:= patch) orelse
(Method =:= put) orelse (Method =:= put) orelse
@ -79,10 +82,12 @@ request_api(Method, Url, QueryParams, AuthOrHeaders, Body) when
end, end,
do_request_api( do_request_api(
Method, Method,
{NewUrl, build_http_header(AuthOrHeaders), "application/json", emqx_json:encode(Body)} {NewUrl, build_http_header(AuthOrHeaders), "application/json", emqx_json:encode(Body)},
Opts
). ).
do_request_api(Method, Request) -> do_request_api(Method, Request, Opts) ->
ReturnBody = maps:get(return_body, Opts, false),
ct:pal("Method: ~p, Request: ~p", [Method, Request]), ct:pal("Method: ~p, Request: ~p", [Method, Request]),
case httpc:request(Method, Request, [], []) of case httpc:request(Method, Request, [], []) of
{error, socket_closed_remotely} -> {error, socket_closed_remotely} ->
@ -91,8 +96,9 @@ do_request_api(Method, Request) ->
Code >= 200 andalso Code =< 299 Code >= 200 andalso Code =< 299
-> ->
{ok, Return}; {ok, Return};
{ok, {Reason, _, _} = Error} -> {ok, {Reason, Headers, Body}} when ReturnBody ->
ct:pal("error: ~p~n", [Error]), {error, {Reason, Headers, Body}};
{ok, {Reason, _Headers, _Body}} ->
{error, Reason} {error, Reason}
end. end.

53
changes/v5.0.1-en.md Normal file
View File

@ -0,0 +1,53 @@
# v5.0.1
5.0.1 is built on [Erlang/OTP 24.2.1-1](https://github.com/emqx/otp/tree/OTP-24.2.1-1). Same as 5.0.0.
5.0.0 (like 4.4.x) had Erlang/OTP version number in the package name.
This is because we wanted to release different flavor packages (on different Elixir/Erlang/OTP platforms).
However the long package names also causes confusion, as users may not know which to choose if there were more than
one presented at the same time.
Going forward, (starting from 5.0.1), packages will be released in both default (short) and flavored (long) package names.
For example: `emqx-5.0.1-otp24.2.1-1-ubuntu20.04-amd64.tar.gz`,
but only the default one is presented to the users: `emqx-5.0.1-ubuntu20.04-amd64.tar.gz`.
In case anyone wants to try a different flavor package, it can be downlowded from the public s3 bucket,
for example:
https://s3.us-west-2.amazonaws.com/packages.emqx/emqx-ce/v5.0.1/emqx-5.0.1-otp24.2.1-1-ubuntu20.04-arm64.tar.gz
Exceptions:
* Windows package is always presented with short name (currently on Erlang/OTP 24.2.1).
* Elixir package name is flavored with both Elixir and Erlang/OTP version numbers,
for example: `emqx-5.0.1-elixir1.13.4-otp24.2.1-1-ubuntu20.04-amd64.tar.gz`
## Enhancements
* Removed management API auth for prometheus scraping endpoint /api/v5/prometheus/stats [#8299](https://github.com/emqx/emqx/pull/8299)
* Added more TCP options for exhook (gRPC) connections. [#8317](https://github.com/emqx/emqx/pull/8317)
* HTTP Servers used for authentication and authorization will now indicate the result via the response body. [#8374](https://github.com/emqx/emqx/pull/8374) [#8377](https://github.com/emqx/emqx/pull/8377)
* Bulk subscribe/unsubscribe APIs [#8356](https://github.com/emqx/emqx/pull/8356)
* Added exclusive subscription [#8315](https://github.com/emqx/emqx/pull/8315)
* Provide authentication counter metrics [#8352](https://github.com/emqx/emqx/pull/8352) [#8375](https://github.com/emqx/emqx/pull/8375)
* Do not allow admin user self-deletion [#8286](https://github.com/emqx/emqx/pull/8286)
* After restart, ensure to copy `cluster-override.conf` from the clustered node which has the greatest `tnxid`. [#8333](https://github.com/emqx/emqx/pull/8333)
## Bug fixes
* A bug fix ported from 4.x: allow deleting subscriptions from `client.subscribe` hookpoint callback result. [#8304](https://github.com/emqx/emqx/pull/8304) [#8347](https://github.com/emqx/emqx/pull/8377)
* Fixed Erlang distribution over TLS [#8309](https://github.com/emqx/emqx/pull/8309)
* Made possible to override authentication configs from environment variables [#8323](https://github.com/emqx/emqx/pull/8309)
* Made authentication passwords in Mnesia database backward compatible to 4.x, so we can support data migration better. [#8351](https://github.com/emqx/emqx/pull/8351)
* Fix plugins upload for rpm/deb installations [#8379](https://github.com/emqx/emqx/pull/8379)
* Sync data/authz/acl.conf and data/certs from clustered nodes after a new node joins the cluster [#8369](https://github.com/emqx/emqx/pull/8369)
* Ensure auto-retry of failed resources [#8371](https://github.com/emqx/emqx/pull/8371)
* Fix the issue that the count of `packets.connack.auth_error` is inaccurate when the client uses a protocol version below MQTT v5.0 to access [#8178](https://github.com/emqx/emqx/pull/8178)
## Others
* Rate limiter interface is hidden so far, it's subject to a UX redesign.
* QUIC library upgraded to 0.0.14.
* Now the default packages will be released withot otp version number in the package name.
* Renamed config exmpale file name in `etc` dir.

11
changes/v5.0.10-en.md Normal file
View File

@ -0,0 +1,11 @@
# v5.0.10
## Enhancements
- Allow clear retained/delayed data when client is banned [#9139](https://github.com/emqx/emqx/pull/9139).
- Update `gen_rpc` library to version 3.0 [#9187](https://github.com/emqx/emqx/pull/9187).
## Bug fixes
- Fix error log message when `mechanism` is missing in authentication config [#8924](https://github.com/emqx/emqx/pull/8924).

11
changes/v5.0.10-zh.md Normal file
View File

@ -0,0 +1,11 @@
# v5.0.10
## 增强
- 支持拉黑客户端并从数据库中删除保留和延迟发布的消息 [#9139](https://github.com/emqx/emqx/pull/9139)。
- 升级 `gen_rpc` 库到 3.0 [#9187](https://github.com/emqx/emqx/pull/9187)。
## Bug fixes
- 优化认认证配置中 `mechanism` 字段缺失情况下的错误日志 [#8924](https://github.com/emqx/emqx/pull/8924)。

18
changes/v5.0.2-en.md Normal file
View File

@ -0,0 +1,18 @@
# v5.0.2
Announcement: EMQX team has decided to stop supporting relup for opensource edition.
Going forward, it will be an enterprise-only feature.
Main reason: relup requires carefully crafted upgrade instructions from ALL previous versions.
For example, 4.3 is now at 4.3.16, we have `4.3.0->4.3.16`, `4.3.1->4.3.16`, ... 16 such upgrade paths in total to maintain.
This had been the biggest obstacle for EMQX team to act agile enough in delivering enhancements and fixes.
## Enhancements
## Bug fixes
* Fixed a typo in `bin/emqx` which affects MacOs release when trying to enable Erlang distribution over TLS [#8398](https://github.com/emqx/emqx/pull/8398)
* Restricted shell was accidentally disabled in 5.0.1, it has been added back. [#8396](https://github.com/emqx/emqx/pull/8396)

12
changes/v5.0.3-en.md Normal file
View File

@ -0,0 +1,12 @@
# v5.0.3
## Bug fixes
* Websocket listener failed to read headers `X-Forwarded-For` and `X-Forwarded-Port` [#8415](https://github.com/emqx/emqx/pull/8415)
* Deleted `cluster_singleton` from MQTT bridge config document. This config is no longer applicable in 5.0 [#8407](https://github.com/emqx/emqx/pull/8407)
* Fix `emqx/emqx:latest` docker image publish to use the Erlang flavor, but not Elixir flavor [#8414](https://github.com/emqx/emqx/pull/8414)
* Changed the `exp` field in JWT auth to be optional rather than required to fix backwards compatability with 4.X releases. [#8425](https://github.com/emqx/emqx/pull/8425)
## Enhancements
* Improve the speed of dashboard's HTTP API routing rule generation, which sometimes causes timeout [#8438](https://github.com/emqx/emqx/pull/8438)

37
changes/v5.0.4-en.md Normal file
View File

@ -0,0 +1,37 @@
# v5.0.4
## Enhancements
* Improve the dashboard listener startup log, the listener name is no longer spliced with port information,
and the colon(:) is no longer displayed when IP is not specified. [#8480](https://github.com/emqx/emqx/pull/8480)
* Remove `/configs/listeners` API, use `/listeners/` instead. [#8485](https://github.com/emqx/emqx/pull/8485)
* Optimize performance of builtin database operations in processes with long message queue [#8439](https://github.com/emqx/emqx/pull/8439)
* Improve authentication tracing. [#8554](https://github.com/emqx/emqx/pull/8554)
* Standardize the '/listeners' and `/gateway/<name>/listeners` API fields.
It will introduce some incompatible updates, see [#8571](https://github.com/emqx/emqx/pull/8571)
* Add option to perform GC on connection process after TLS/SSL handshake is performed. [#8637](https://github.com/emqx/emqx/pull/8637)
## Bug fixes
* The `data/configs/cluster-override.conf` is cleared to 0KB if `hocon_pp:do/2` failed [commits/71f64251](https://github.com/emqx/emqx/pull/8443/commits/71f642518a683cc91a32fd542aafaac6ef915720)
* Improve the health_check for webhooks.
Prior to this change, the webhook only checks the connectivity of the TCP port using `gen_tcp:connect/2`, so
if it's a HTTPs server, we didn't check if TLS handshake was successful.
[commits/6b45d2ea](https://github.com/emqx/emqx/commit/6b45d2ea9fde6d3b4a5b007f7a8c5a1c573d141e)
* The `created_at` field of rules is missing after emqx restarts. [commits/5fc09e6b](https://github.com/emqx/emqx/commit/5fc09e6b950c340243d7be627a0ce1700691221c)
* The rule engine's jq function now works even when the path to the EMQX install dir contains spaces [jq#35](https://github.com/emqx/jq/pull/35) [#8455](https://github.com/emqx/emqx/pull/8455)
* Avoid applying any ACL checks on superusers [#8452](https://github.com/emqx/emqx/pull/8452)
* Fix statistics related system topic name error
* Fix AuthN JWKS SSL schema. Using schema in `emqx_schema`. [#8458](https://github.com/emqx/emqx/pull/8458)
* `sentinel` field should be required when AuthN/AuthZ Redis using sentinel mode. [#8458](https://github.com/emqx/emqx/pull/8458)
* Fix bad swagger format. [#8517](https://github.com/emqx/emqx/pull/8517)
* Fix `chars_limit` is not working when `formatter` is `json`. [#8518](http://github.com/emqx/emqx/pull/8518)
* Ensuring that exhook dispatches the client events are sequential. [#8530](https://github.com/emqx/emqx/pull/8530)
* Avoid using RocksDB backend for persistent sessions when such backend is unavailable. [#8528](https://github.com/emqx/emqx/pull/8528)
* Fix AuthN `cert_subject` and `cert_common_name` placeholder rendering failure. [#8531](https://github.com/emqx/emqx/pull/8531)
* Support listen on an IPv6 address, e.g: [::1]:1883 or ::1:1883. [#8547](https://github.com/emqx/emqx/pull/8547)
* GET '/rules' support for pagination and fuzzy search. [#8472](https://github.com/emqx/emqx/pull/8472)
**‼️ Note** : The previous API only returns array: `[RuleObj1,RuleObj2]`, after updating, it will become
`{"data": [RuleObj1,RuleObj2], "meta":{"count":2, "limit":100, "page":1}`,
which will carry the paging meta information.
* Fix the issue that webhook leaks TCP connections. [ehttpc#34](https://github.com/emqx/ehttpc/pull/34), [#8580](https://github.com/emqx/emqx/pull/8580)

17
changes/v5.0.5-en.md Normal file
View File

@ -0,0 +1,17 @@
# v5.0.5
## Enhancements
* Add `bootstrap_users_file` configuration to add default Dashboard username list, which is only added when EMQX is first started.
* The license is now copied to all nodes in the cluster when it's reloaded. [#8598](https://github.com/emqx/emqx/pull/8598)
* Added a HTTP API to manage licenses. [#8610](https://github.com/emqx/emqx/pull/8610)
* Updated `/nodes` API node_status from `Running/Stopped` to `running/stopped`. [#8642](https://github.com/emqx/emqx/pull/8642)
* Improve handling of placeholder interpolation errors [#8635](https://github.com/emqx/emqx/pull/8635)
* Better logging on unknown object IDs. [#8670](https://github.com/emqx/emqx/pull/8670)
* The bind option support `:1883` style. [#8758](https://github.com/emqx/emqx/pull/8758)
## Bug fixes
* Allow changing the license type from key to file (and vice-versa). [#8598](https://github.com/emqx/emqx/pull/8598)
* Add back http connector config keys `max_retries` `retry_interval` as deprecated fields [#8672](https://github.com/emqx/emqx/issues/8672)
This caused upgrade failure in 5.0.4, because it would fail to boot on configs created from older version.

5
changes/v5.0.6-en.md Normal file
View File

@ -0,0 +1,5 @@
# v5.0.6
## Bug fixes
* Upgrade Dashboard version to fix an issue where the node status was not displayed correctly. [#8771](https://github.com/emqx/emqx/pull/8771)

12
changes/v5.0.7-en.md Normal file
View File

@ -0,0 +1,12 @@
# v5.0.7
## Enhancements
* Do not auto-populate default SSL cipher suites, so that the configs are less bloated. [#8769](https://github.com/emqx/emqx/pull/8769)
## Bug fixes
* Remove `will_msg` (not used) field from the client API. [#8721](https://github.com/emqx/emqx/pull/8721)
* Fix `$queue` topic name error in management API return. [#8728](https://github.com/emqx/emqx/pull/8728)
* Fix race condition which may cause `client.connected` and `client.disconnected` out of order. [#8625](https://github.com/emqx/emqx/pull/8625)
* Fix quic listener default idle timeout's type. [#8826](https://github.com/emqx/emqx/pull/8826)

28
changes/v5.0.8-en.md Normal file
View File

@ -0,0 +1,28 @@
# v5.0.8
## Enhancements
* Print a warning message when boot with the default (insecure) Erlang cookie. [#8905](https://github.com/emqx/emqx/pull/8905)
* Change the `/gateway` API path to plural form. [#8823](https://github.com/emqx/emqx/pull/8823)
* Don't allow updating config items when they already exist in `local-override.conf`. [#8851](https://github.com/emqx/emqx/pull/8851)
* Remove `node.etc_dir` from emqx.conf, because it is never used.
Also allow user to customize the logging directory [#8892](https://github.com/emqx/emqx/pull/8892)
* Added a new API `POST /listeners` for creating listener. [#8876](https://github.com/emqx/emqx/pull/8876)
* Close ExProto client process immediately if it's keepalive timeouted. [#8866](https://github.com/emqx/emqx/pull/8866)
* Upgrade grpc-erl driver to 0.6.7 to support batch operation in sending stream. [#8866](https://github.com/emqx/emqx/pull/8866)
## Bug fixes
* Fix exhook `client.authorize` never being execauted. [#8780](https://github.com/emqx/emqx/pull/8780)
* Fix JWT plugin don't support non-integer timestamp claims. [#8867](https://github.com/emqx/emqx/pull/8867)
* Avoid publishing will message when client fails to auhtenticate. [#8887](https://github.com/emqx/emqx/pull/8887)
* Speed up dispatching of shared subscription messages in a cluster [#8893](https://github.com/emqx/emqx/pull/8893)
* Fix the extra / prefix when CoAP gateway parsing client topics. [#8658](https://github.com/emqx/emqx/pull/8658)
* Speed up updating the configuration, When some nodes in the cluster are down. [#8857](https://github.com/emqx/emqx/pull/8857)
* Fix delayed publish inaccurate caused by os time change. [#8926](https://github.com/emqx/emqx/pull/8926)
* Fix that EMQX can't start when the retainer is disabled [#8911](https://github.com/emqx/emqx/pull/8911)
* Fix that redis authn will deny the unknown users [#8934](https://github.com/emqx/emqx/pull/8934)
* Fix ExProto UDP client keepalive checking error.
This causes the clients to not expire as long as a new UDP packet arrives [#8866](https://github.com/emqx/emqx/pull/8866)
* Fix that MQTT Bridge message payload could be empty string. [#8949](https://github.com/emqx/emqx/pull/8949)

34
changes/v5.0.9-en.md Normal file
View File

@ -0,0 +1,34 @@
# v5.0.9
## Enhancements
- Add `cert_common_name` and `cert_subject` placeholder support for authz_http and authz_mongo [#8973](https://github.com/emqx/emqx/pull/8973).
- Use milliseconds internally in emqx_delayed to store the publish time, improving precision [#9060](https://github.com/emqx/emqx/pull/9060).
- More rigorous checking of flapping to improve stability of the system [#9136](https://github.com/emqx/emqx/pull/9136).
- No message(s) echo for the message publish APIs [#9155](https://github.com/emqx/emqx/pull/9155).
Prior to this fix, the message publish APIs (`api/v5/publish` and `api/v5/publish/bulk`) echos the message back to the client in HTTP body.
This change fixed it to only send back the message ID.
## Bug fixes
- Check ACLs for last will testament topic before publishing the message [#8930](https://github.com/emqx/emqx/pull/8930).
- Fix GET /listeners API crash when some nodes (in a cluster) is still loading the configs [#9002](https://github.com/emqx/emqx/pull/9002).
- Fix empty variable interpolation in authentication and authorization [#8963](https://github.com/emqx/emqx/pull/8963).
Placeholders for undefined variables are rendered now as empty strings and do not cause errors anymore.
- Fix the latency statistics error of the slow subscription stats [#8986](https://github.com/emqx/emqx/pull/8986).
Prior to this change when `stats_type` is `internal` or `response`, the begin time stamp was taken at wrong precision.
- Fix shared subscription message re-dispatches [#9104](https://github.com/emqx/emqx/pull/9104).
- When discarding QoS 2 inflight messages, there were excessive logs
- For wildcard deliveries, the re-dispatch used the wrong topic (the publishing topic,
but not the subscribing topic), caused messages to be lost when dispatching.
- Upgrade http client `gun` from 1.3.7 to [1.3.9](https://github.com/emqx/gun/tree/1.3.9)
Prior to this fix, long-lived HTTPS connections for HTTP auth or webhook integrations
may stall indefinitely, causing massive timeouts for HTTP requests.

28
changes/v5.0.9-zh.md Normal file
View File

@ -0,0 +1,28 @@
# v5.0.9
## 增强
- 为 `authz_http` 和 `authz_mongo` 增加了 `cert_common_name` 和 `cert_subject` 两个占位符 [#8973](https://github.com/emqx/emqx/pull/8973)。
- 统一使用 Erlang 虚拟机的时间，而不是系统时间，可以避免系统时间修改后导致的延迟发布不准确问题 [#9060](https://github.com/emqx/emqx/pull/9060)。
- 更严格的 flapping 检测，认证失败等也会进行计数 [#9136](https://github.com/emqx/emqx/pull/9136)。
## 修复
- 遗嘱消息发布前进行 ACL 检查 [#8930](https://github.com/emqx/emqx/pull/8930)。
- 在集群环境下，当有节点还没有完全初始化好配置时，`GET /listeners` 可能会返回 HTTP 500 的错误 [#9002](https://github.com/emqx/emqx/pull/9002)。
- 认证和鉴权的占位符替换中，如果没有找到匹配的值，使用空字符串代替，而不是抛出一异常 [#8963](https://github.com/emqx/emqx/pull/8963)。
- 慢订阅统计中时间单位用错的问题 [#8986](https://github.com/emqx/emqx/pull/8986)。
当统计类型（`stats_type`）是 `internal` 或者 `response` 时，起始时间戳的精度使用错误。
- 共享订阅消息重新派发 [#9104](https://github.com/emqx/emqx/pull/9104)。
- 当 QoS 2 的 inflight 消息被丢弃时，产生了大量的 warning 日志，修复后不再打印。
- 通配符订阅的共享订阅消息在重新派发时，使用了消息发布时的主题，而不是订阅的通配符主题选择
订阅组中的其他成员，导致转发失败。
- HTTP 客户端 （`gun`） 从 1.3.7 升级到 [1.3.9](https://github.com/emqx/gun/tree/1.3.9)。
此次修复前，HTTP 认证和 webhook 等使用 HTTPS 客户端长连接的后端可能会进入一个无限等待状态，导致大量超时发生。

4
deploy/charts/emqx/Chart.yaml
View File

@ -14,8 +14,8 @@ type: application
# This is the chart version. This version number should be incremented each time you make changes # This is the chart version. This version number should be incremented each time you make changes
# to the chart and its templates, including the app version. # to the chart and its templates, including the app version.
version: 5.0.8 version: 5.0.9
# This is the version number of the application being deployed. This version number should be # This is the version number of the application being deployed. This version number should be
# incremented each time you make changes to the application. # incremented each time you make changes to the application.
appVersion: 5.0.8 appVersion: 5.0.9

2
mix.exs
View File

@ -71,7 +71,7 @@ defmodule EMQXUmbrella.MixProject do
{:esasl, github: "emqx/esasl", tag: "0.2.0"}, {:esasl, github: "emqx/esasl", tag: "0.2.0"},
{:jose, github: "potatosalad/erlang-jose", tag: "1.11.2"}, {:jose, github: "potatosalad/erlang-jose", tag: "1.11.2"},
# in conflict by ehttpc and emqtt # in conflict by ehttpc and emqtt
{:gun, github: "emqx/gun", tag: "1.3.7", override: true}, {:gun, github: "emqx/gun", tag: "1.3.9", override: true},
# in conflict by emqx_connectior and system_monitor # in conflict by emqx_connectior and system_monitor
{:epgsql, github: "emqx/epgsql", tag: "4.7-emqx.2", override: true}, {:epgsql, github: "emqx/epgsql", tag: "4.7-emqx.2", override: true},
# in conflict by mongodb and eredis_cluster # in conflict by mongodb and eredis_cluster

2
rebar.config
View File

@ -48,7 +48,7 @@
, {redbug, "2.0.7"} , {redbug, "2.0.7"}
, {gpb, "4.19.5"} %% gpb only used to build, but not for release, pin it here to avoid fetching a wrong version due to rebar plugins scattered in all the deps , {gpb, "4.19.5"} %% gpb only used to build, but not for release, pin it here to avoid fetching a wrong version due to rebar plugins scattered in all the deps
, {typerefl, {git, "https://github.com/ieQu1/typerefl", {tag, "0.9.1"}}} , {typerefl, {git, "https://github.com/ieQu1/typerefl", {tag, "0.9.1"}}}
, {gun, {git, "https://github.com/emqx/gun", {tag, "1.3.7"}}} , {gun, {git, "https://github.com/emqx/gun", {tag, "1.3.9"}}}
, {ehttpc, {git, "https://github.com/emqx/ehttpc", {tag, "0.3.0"}}} , {ehttpc, {git, "https://github.com/emqx/ehttpc", {tag, "0.3.0"}}}
, {gproc, {git, "https://github.com/uwiger/gproc", {tag, "0.8.0"}}} , {gproc, {git, "https://github.com/uwiger/gproc", {tag, "0.8.0"}}}
, {jiffy, {git, "https://github.com/emqx/jiffy", {tag, "1.0.5"}}} , {jiffy, {git, "https://github.com/emqx/jiffy", {tag, "1.0.5"}}}