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

Merge pull request #10854 from terry-xiaoyu/sync-code-from-ee 

feat: configurable TCP keepalive
This commit is contained in:
Xinyu Liu 2023-05-29 19:49:48 +08:00 committed by GitHub
parent 3e7ec9d008 797c422221
commit abbea11a44
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 229 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
changes/v4.4.19-en.md
View File

@ -1,5 +1,9 @@
# v4.4.19
- Added support for TCP keep-alive in MQTT/TCP and MQTT/SSL listeners [#10854](https://github.com/emqx/emqx/pull/10854).
A new configuration option has been added: `zone.<zone-name>.tcp_keepalive = Idle,Interval,Probes`. Users can enable the TCP layer's Keep Alive feature and specify time parameters using this configuration. This configuration is only effective on Linux and MacOS systems.
## Enhancements
- Improving error logs related to Proxy Protocol [emqx/esockd#177](https://github.com/emqx/esockd/pull/177).

6
changes/v4.4.19-zh.md
View File

@ -2,6 +2,10 @@
## 增强
- 为 MQTT/TCP 和 MQTT/SSL 监听器增加 TCP Keep Alive 的支持 [#10854](https://github.com/emqx/emqx/pull/10854)。
现在增加了一个配置项：`zone.<zone-name>.tcp_keepalive = Idle,Interval,Probes`，用户可以通过此配置来启用 TCP 层的 Keep Alive 功能并指定时间参数。此配置仅在 Linux 和 MacOS 系统上生效。
- 改进 Proxy Protocol 相关的错误日志 [emqx/esockd#177](https://github.com/emqx/esockd/pull/177)。
改进之前的日志样例:
@ -18,7 +22,7 @@
```
- 增加了一个新功能，为 TLS 监听器启用部分证书链验证[#10553](https://github.com/emqx/emqx/pull/10553)。
- 增加了一个新功能，为 TLS 监听器启用客户端证书扩展密钥使用验证 [#10669](https://github.com/emqx/emqx/pull/10669)。
## 修复

46
etc/emqx.conf
View File

@ -1072,6 +1072,29 @@ zone.external.strict_mode = false
## Value: String
## zone.external.response_information = example
## Enable TCP keepalive for MQTT connections over TCP or SSL.
##
## The value is three comma separated numbers in the format of
## 'Idle,Interval,Probes'
##
## Idle: The number of seconds a connection needs to be idle before
## the server begins to send out keep-alive probes (Linux default 7200).
##
## Interval: The number of seconds between TCP keep-alive probes (Linux default 75).
##
## Probes: The maximum number of TCP keep-alive probes to send before
## giving up and killing the connection if no response is
## obtained from the other end (Linux default 9).
##
## For example "240,30,5" means: EMQX should start sending TCP keepalive probes
## after the connection is in idel for 240 seconds,
## and the probes are sent every 30 seconds until a response is received from the MQTT
## client, if it misses 5 consecutive responses, EMQX should close the connection.
##
## Value: string
## Default: none
## zone.external.tcp_keepalive = none
##--------------------------------------------------------------------
## Internal Zone
@ -1177,6 +1200,29 @@ zone.internal.strict_mode = false
## Value: true | false
zone.internal.bypass_auth_plugins = true
## Enable TCP keepalive for MQTT connections over TCP or SSL.
##
## The value is three comma separated numbers in the format of
## 'Idle,Interval,Probes'
##
## Idle: The number of seconds a connection needs to be idle before
## the server begins to send out keep-alive probes (Linux default 7200).
##
## Interval: The number of seconds between TCP keep-alive probes (Linux default 75).
##
## Probes: The maximum number of TCP keep-alive probes to send before
## giving up and killing the connection if no response is
## obtained from the other end (Linux default 9).
##
## For example "240,30,5" means: EMQX should start sending TCP keepalive probes
## after the connection is in idel for 240 seconds,
## and the probes are sent every 30 seconds until a response is received from the MQTT
## client, if it misses 5 consecutive responses, EMQX should close the connection.
##
## Value: string
## Default: none
## zone.internal.tcp_keepalive = none
## CONFIG_SECTION_END=zones ====================================================
## CONFIG_SECTION_BGN=listeners ================================================

25
priv/emqx.schema
View File

@ -166,11 +166,11 @@
{Ip, Port}
end,
Options = fun(static) ->
[{seeds, [list_to_atom(S) || S <- string:tokens(cuttlefish:conf_get("cluster.static.seeds", Conf, ""), ",")]}];
[{seeds, [list_to_atom(S) || S <- string:tokens(cuttlefish:conf_get("cluster.static.seeds", Conf, ""), ", ")]}];
(mcast) ->
{ok, Addr} = inet:parse_address(cuttlefish:conf_get("cluster.mcast.addr", Conf)),
{ok, Iface} = inet:parse_address(cuttlefish:conf_get("cluster.mcast.iface", Conf)),
Ports = [list_to_integer(S) || S <- string:tokens(cuttlefish:conf_get("cluster.mcast.ports", Conf), ",")],
Ports = [list_to_integer(S) || S <- string:tokens(cuttlefish:conf_get("cluster.mcast.ports", Conf), ", ")],
[{addr, Addr}, {ports, Ports}, {iface, Iface},
{ttl, cuttlefish:conf_get("cluster.mcast.ttl", Conf, 1)},
{loop, cuttlefish:conf_get("cluster.mcast.loop", Conf, true)}];
@ -185,7 +185,7 @@
{list_to_atom(Name), Value}
end, Options)
end,
[{server, string:tokens(cuttlefish:conf_get("cluster.etcd.server", Conf), ",")},
[{server, string:tokens(cuttlefish:conf_get("cluster.etcd.server", Conf), ", ")},
{prefix, cuttlefish:conf_get("cluster.etcd.prefix", Conf, "emqcl")},
{node_ttl, cuttlefish:conf_get("cluster.etcd.node_ttl", Conf, 60)},
{ssl_options, SslOpts(Conf)}];
@ -1286,6 +1286,12 @@ end}.
{datatype, {enum, [true, false]}}
]}.
%% @doc Set TCP-keepalive options.
{mapping, "zone.$name.tcp_keepalive", "emqx.zones", [
{default, "none"},
{datatype, string}
]}.
{translation, "emqx.zones", fun(Conf) ->
Ratelimit = fun(Val) ->
[L, D] = string:tokens(Val, ", "),
@ -1371,6 +1377,15 @@ end}.
{quota, {conn_messages_routing, Ratelimit(Val)}};
(["quota", "overall_messages_routing"], Val) ->
{quota, {overall_messages_routing, Ratelimit(Val)}};
(["tcp_keepalive"], Val) ->
V = case Val of
"none" ->
false;
_ ->
[Idle, Interval, Probes] = string:tokens(Val, ", "),
{list_to_integer(Idle), list_to_integer(Interval), list_to_integer(Probes)}
end,
{tcp_keepalive, V};
([Opt], Val) ->
{list_to_atom(Opt), Val}
end,
@ -2227,7 +2242,7 @@ end}.
end,
CheckOrigin = fun(S) ->
Origins = string:tokens(S, ","),
Origins = string:tokens(S, ", "),
[ list_to_binary(string:trim(O)) || O <- Origins]
end,
@ -2873,7 +2888,7 @@ end}.
]}.
{translation, "emqx.alarm", fun(Conf) ->
[{actions, [list_to_atom(Action) || Action <- string:tokens(cuttlefish:conf_get("alarm.actions", Conf), ",")]},
[{actions, [list_to_atom(Action) || Action <- string:tokens(cuttlefish:conf_get("alarm.actions", Conf), ", ")]},
{size_limit, cuttlefish:conf_get("alarm.size_limit", Conf)},
{validity_period, cuttlefish:conf_get("alarm.validity_period", Conf)}]
end}.

1
scripts/macos-sign-binaries.sh
View File

@ -68,6 +68,7 @@ for f in \
sasl_auth.so \
snappyer.so \
odbcserver \
ezstd_nif.so \
; do
find "${REL_DIR}"/lib/ -name "$f" -exec codesign -s "${APPLE_DEVELOPER_IDENTITY}" -f --verbose=4 --timestamp --options=runtime {} \;
done

13
scripts/relup-base-packages.sh
View File

@ -64,17 +64,24 @@ mkdir -p _upgrade_base
pushd _upgrade_base
otp_vsn_for() {
../scripts/relup-base-vsns.escript otp-vsn-for "${1#[e|v]}" ../data/relup-paths.eterm
case "$PROFILE" in
*-ee-*)
../scripts/relup-base-vsns.escript otp-vsn-for "${1#[e|v]}" ../data/relup-paths-ee.eterm
;;
*)
../scripts/relup-base-vsns.escript otp-vsn-for "${1#[e|v]}" ../data/relup-paths.eterm
;;
esac
}
for tag in $(../scripts/relup-base-vsns.sh $EDITION | xargs echo -n); do
filename="$PROFILE-${tag#[e|v]}-otp$(otp_vsn_for "$tag")-$SYSTEM-$ARCH.zip"
url="https://packages.emqx.io/$DIR/$tag/$filename"
echo "downloading base package from ${url} ..."
if [ -f "$filename" ]; then
echo "file $filename already downloaded; skikpped"
echo "file $filename already downloaded; skipped"
continue
fi
curl -L -I -m 10 -o /dev/null -s -w "%{http_code}" "${url}" | grep -q -oE "^[23]+"
echo "downloading base package from ${url} ..."
curl -L -o "${filename}" "${url}"
if [ "$SYSTEM" != "centos6" ]; then

16
scripts/split-config.escript
View File

@ -57,7 +57,15 @@ parse_sections([Line | Lines], Parse, Section, Sections) ->
dump_sections([]) -> ok;
dump_sections([{Name, Lines0} | Rest]) ->
Filename = filename:join(["etc", iolist_to_binary([Name, ".conf.seg"])]),
Lines = [[L, "\n"] || L <- Lines0],
ok = file:write_file(Filename, Lines),
dump_sections(Rest).
case is_skipped(Name) of
true ->
dump_sections(Rest);
false ->
Filename = filename:join(["etc", iolist_to_binary([Name, ".conf.seg"])]),
Lines = [[L, "\n"] || L <- Lines0],
ok = file:write_file(Filename, Lines),
dump_sections(Rest)
end.
is_skipped(Name) ->
Name =:= <<"modules">>.

114
src/emqx.appup.src
View File

@ -2,7 +2,8 @@
%% Unless you know what you are doing, DO NOT edit manually!!
{VSN,
[{"4.4.18",
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_cm,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
@ -12,7 +13,8 @@
{load_module,emqx_relup,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.4.17",
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_cm,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
@ -23,7 +25,8 @@
{load_module,emqx_app,brutal_purge,soft_purge,[]},
{load_module,emqx_plugins,brutal_purge,soft_purge,[]}]},
{"4.4.16",
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_cm,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
@ -37,7 +40,8 @@
{load_module,emqx_plugins,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.4.15",
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_cm,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
@ -53,7 +57,8 @@
{load_module,emqx_relup,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.4.14",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_cm,brutal_purge,soft_purge,[]},
{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
@ -74,7 +79,8 @@
{load_module,emqx_app,brutal_purge,soft_purge,[]},
{load_module,emqx_rule_actions_trans,brutal_purge,soft_purge,[]}]},
{"4.4.13",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{add_module,emqx_const_v2},
@ -95,7 +101,8 @@
{load_module,emqx_relup,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.4.12",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{add_module,emqx_const_v2},
@ -116,7 +123,8 @@
{load_module,emqx_relup,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.4.11",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{add_module,emqx_const_v2},
@ -139,7 +147,8 @@
{load_module,emqx_session,brutal_purge,soft_purge,[]},
{load_module,emqx_channel,brutal_purge,soft_purge,[]}]},
{"4.4.10",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{add_module,emqx_const_v2},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
@ -170,7 +179,8 @@
{apply,{application,set_env,
[gen_rpc,insecure_auth_fallback_allowed,true]}}]},
{"4.4.9",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{add_module,emqx_const_v2},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
@ -206,7 +216,8 @@
{apply,{application,set_env,
[gen_rpc,insecure_auth_fallback_allowed,true]}}]},
{"4.4.8",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{add_module,emqx_const_v2},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
@ -243,7 +254,8 @@
{apply,{application,set_env,
[gen_rpc,insecure_auth_fallback_allowed,true]}}]},
{"4.4.7",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{add_module,emqx_const_v2},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
@ -280,7 +292,8 @@
{apply,{application,set_env,
[gen_rpc,insecure_auth_fallback_allowed,true]}}]},
{"4.4.6",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{add_module,emqx_const_v2},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
@ -317,7 +330,8 @@
{apply,{application,set_env,
[gen_rpc,insecure_auth_fallback_allowed,true]}}]},
{"4.4.5",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{add_module,emqx_const_v2},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
@ -356,7 +370,8 @@
{apply,{application,set_env,
[gen_rpc,insecure_auth_fallback_allowed,true]}}]},
{"4.4.4",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{add_module,emqx_const_v2},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
@ -401,7 +416,8 @@
{apply,{application,set_env,
[gen_rpc,insecure_auth_fallback_allowed,true]}}]},
{"4.4.3",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{add_module,emqx_const_v2},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
@ -452,7 +468,8 @@
{apply,{application,set_env,
[gen_rpc,insecure_auth_fallback_allowed,true]}}]},
{"4.4.2",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{add_module,emqx_const_v2},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
@ -504,7 +521,8 @@
{apply,{application,set_env,
[gen_rpc,insecure_auth_fallback_allowed,true]}}]},
{"4.4.1",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{add_module,emqx_const_v2},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
@ -560,7 +578,8 @@
{apply,{application,set_env,
[gen_rpc,insecure_auth_fallback_allowed,true]}}]},
{"4.4.0",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{add_module,emqx_const_v2},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
@ -619,7 +638,8 @@
[gen_rpc,insecure_auth_fallback_allowed,true]}}]},
{<<".*">>,[]}],
[{"4.4.18",
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_cm,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
@ -628,7 +648,8 @@
{load_module,emqx_relup,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.4.17",
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_cm,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
@ -638,7 +659,8 @@
{load_module,emqx_app,brutal_purge,soft_purge,[]},
{load_module,emqx_plugins,brutal_purge,soft_purge,[]}]},
{"4.4.16",
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_cm,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
@ -651,7 +673,8 @@
{load_module,emqx_relup,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.4.15",
[{load_module,emqx_connection,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_connection,brutal_purge,soft_purge,[]},
{load_module,emqx_cm,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
@ -666,7 +689,8 @@
{load_module,emqx_relup,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.4.14",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_cm,brutal_purge,soft_purge,[]},
{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
@ -686,7 +710,8 @@
{load_module,emqx_app,brutal_purge,soft_purge,[]},
{load_module,emqx_rule_actions_trans,brutal_purge,soft_purge,[]}]},
{"4.4.13",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{load_module,emqx_channel,brutal_purge,soft_purge,[]},
@ -706,7 +731,8 @@
{load_module,emqx_relup,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.4.12",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{load_module,emqx_channel,brutal_purge,soft_purge,[]},
@ -726,7 +752,8 @@
{load_module,emqx_relup,brutal_purge,soft_purge,[]},
{load_module,emqx_app,brutal_purge,soft_purge,[]}]},
{"4.4.11",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_listeners,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
@ -748,7 +775,8 @@
{load_module,emqx_session,brutal_purge,soft_purge,[]},
{delete_module,emqx_cover}]},
{"4.4.10",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
{load_module,emqx_vm,brutal_purge,soft_purge,[]},
@ -775,7 +803,8 @@
{delete_module,emqx_crl_cache},
{delete_module,emqx_ocsp_cache}]},
{"4.4.9",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
{load_module,emqx_vm,brutal_purge,soft_purge,[]},
@ -806,7 +835,8 @@
{delete_module,emqx_crl_cache},
{delete_module,emqx_ocsp_cache}]},
{"4.4.8",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
{load_module,emqx_vm,brutal_purge,soft_purge,[]},
@ -838,7 +868,8 @@
{delete_module,emqx_crl_cache},
{delete_module,emqx_ocsp_cache}]},
{"4.4.7",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
{load_module,emqx_vm,brutal_purge,soft_purge,[]},
@ -870,7 +901,8 @@
{delete_module,emqx_crl_cache},
{delete_module,emqx_ocsp_cache}]},
{"4.4.6",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
{load_module,emqx_vm,brutal_purge,soft_purge,[]},
@ -902,7 +934,8 @@
{delete_module,emqx_crl_cache},
{delete_module,emqx_ocsp_cache}]},
{"4.4.5",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
{load_module,emqx_vm,brutal_purge,soft_purge,[]},
@ -936,7 +969,8 @@
{delete_module,emqx_crl_cache},
{delete_module,emqx_ocsp_cache}]},
{"4.4.4",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
{load_module,emqx_vm,brutal_purge,soft_purge,[]},
@ -976,7 +1010,8 @@
{delete_module,emqx_crl_cache},
{delete_module,emqx_ocsp_cache}]},
{"4.4.3",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
{load_module,emqx_vm,brutal_purge,soft_purge,[]},
@ -1021,7 +1056,8 @@
{delete_module,emqx_crl_cache},
{delete_module,emqx_ocsp_cache}]},
{"4.4.2",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
{load_module,emqx_vm,brutal_purge,soft_purge,[]},
@ -1067,7 +1103,8 @@
{delete_module,emqx_crl_cache},
{delete_module,emqx_ocsp_cache}]},
{"4.4.1",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
{load_module,emqx_vm,brutal_purge,soft_purge,[]},
@ -1117,7 +1154,8 @@
{delete_module,emqx_crl_cache},
{delete_module,emqx_ocsp_cache}]},
{"4.4.0",
[{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
[{load_module,emqx_zone,brutal_purge,soft_purge,[]},
{load_module,emqx_cm_locker,brutal_purge,soft_purge,[]},
{load_module,emqx_tls_lib,brutal_purge,soft_purge,[]},
{load_module,emqx_pool,brutal_purge,soft_purge,[]},
{load_module,emqx_vm,brutal_purge,soft_purge,[]},

33
src/emqx_connection.erl
View File

@ -40,7 +40,7 @@
]).
-export([ async_set_keepalive/3
, async_set_keepalive/4
, async_set_keepalive/5
, async_set_socket_options/2
]).
@ -226,15 +226,24 @@ stats(#state{transport = Transport,
%% NOTE: This API sets TCP socket options, which has nothing to do with
%% the MQTT layer's keepalive (PINGREQ and PINGRESP).
async_set_keepalive(Idle, Interval, Probes) ->
async_set_keepalive(self(), Idle, Interval, Probes).
async_set_keepalive(os:type(), self(), Idle, Interval, Probes).
async_set_keepalive(Pid, Idle, Interval, Probes) ->
async_set_keepalive({unix, linux}, Pid, Idle, Interval, Probes) ->
Options = [ {keepalive, true}
, {raw, 6, 4, <<Idle:32/native>>}
, {raw, 6, 5, <<Interval:32/native>>}
, {raw, 6, 6, <<Probes:32/native>>}
],
async_set_socket_options(Pid, Options).
async_set_socket_options(Pid, Options);
async_set_keepalive({unix, darwin}, Pid, Idle, Interval, Probes) ->
Options = [ {keepalive, true}
, {raw, 6, 16#10, <<Idle:32/native>>}
, {raw, 6, 16#101, <<Interval:32/native>>}
, {raw, 6, 16#102, <<Probes:32/native>>}
],
async_set_socket_options(Pid, Options);
async_set_keepalive(_Unsupported, _Pid, _Idle, _Interval, _Probes) ->
ok.
%% @doc Set custom socket options.
%% This API is made async because the call might be originated from
@ -291,6 +300,12 @@ init_state(Transport, Socket, Options) ->
StatsTimer = emqx_zone:stats_timer(Zone),
IdleTimeout = emqx_zone:idle_timeout(Zone),
IdleTimer = start_timer(IdleTimeout, idle_timeout),
case emqx_zone:tcp_keepalive(Zone) of
false ->
ok;
{Idle, Interval, Probes} ->
ok = async_set_keepalive(Idle, Interval, Probes)
end,
#state{transport = Transport,
socket = Socket,
peername = Peername,
@ -769,8 +784,14 @@ handle_cast({async_set_socket_options, Opts},
socket = Socket
}) ->
case Transport:setopts(Socket, Opts) of
ok -> ?tp(info, "custom_socket_options_successfully", #{opts => Opts});
Err -> ?tp(error, "failed_to_set_custom_socket_optionn", #{reason => Err})
ok ->
?tp(debug, "custom_socket_options_successfully", #{opts => Opts});
{error, einval} ->
%% socket is already closed, ignore this error
ok;
Err ->
%% other errors
?tp(error, "failed_to_set_custom_socket_optionn", #{reason => Err})
end,
State;
handle_cast(Req, State) ->

11
src/emqx_zone.erl
View File

@ -79,6 +79,7 @@
, force_shutdown_policy/1
, response_information/1
, quota_policy/1
, tcp_keepalive/1
]).
-export([ init_gc_state/1
@ -223,6 +224,16 @@ response_information(Zone) ->
quota_policy(Zone) ->
get_env(Zone, quota, []).
-spec tcp_keepalive(zone()) -> false | {integer(), integer(), integer()}.
tcp_keepalive(Zone) ->
case get_env(Zone, tcp_keepalive, false) of
{_, _, _} = V ->
V;
_ ->
%% failed to parse
false
end.
%%--------------------------------------------------------------------
%% APIs
%%--------------------------------------------------------------------

26
test/emqx_mqtt_SUITE.erl
View File

@ -183,13 +183,15 @@ t_async_set_keepalive('end', _Config) ->
t_async_set_keepalive(_) ->
case os:type() of
{unix, darwin} ->
%% Mac OSX don't support the feature
ok;
do_async_set_keepalive(16#10, 16#101, 16#102);
{unix, linux} ->
do_async_set_keepalive(4, 5, 6);
_ ->
do_async_set_keepalive()
%% don't support the feature on other OS
ok
end.
do_async_set_keepalive() ->
do_async_set_keepalive(OptKeepIdle, OptKeepInterval, OptKeepCount) ->
ClientID = <<"client-tcp-keepalive">>,
{ok, Client} = emqtt:start_link([{host, "localhost"},
{proto_ver,v5},
@ -203,17 +205,17 @@ do_async_set_keepalive() ->
Transport = maps:get(transport, State),
Socket = maps:get(socket, State),
?assert(is_port(Socket)),
Opts = [{raw, 6, 4, 4}, {raw, 6, 5, 4}, {raw, 6, 6, 4}],
{ok, [ {raw, 6, 4, <<Idle:32/native>>}
, {raw, 6, 5, <<Interval:32/native>>}
, {raw, 6, 6, <<Probes:32/native>>}
Opts = [{raw, 6, OptKeepIdle, 4}, {raw, 6, OptKeepInterval, 4}, {raw, 6, OptKeepCount, 4}],
{ok, [ {raw, 6, OptKeepIdle, <<Idle:32/native>>}
, {raw, 6, OptKeepInterval, <<Interval:32/native>>}
, {raw, 6, OptKeepCount, <<Probes:32/native>>}
]} = Transport:getopts(Socket, Opts),
ct:pal("Idle=~p, Interval=~p, Probes=~p", [Idle, Interval, Probes]),
emqx_connection:async_set_keepalive(Pid, Idle + 1, Interval + 1, Probes + 1),
emqx_connection:async_set_keepalive(os:type(), Pid, Idle + 1, Interval + 1, Probes + 1),
{ok, _} = ?block_until(#{?snk_kind := "custom_socket_options_successfully"}, 1000),
{ok, [ {raw, 6, 4, <<NewIdle:32/native>>}
, {raw, 6, 5, <<NewInterval:32/native>>}
, {raw, 6, 6, <<NewProbes:32/native>>}
{ok, [ {raw, 6, OptKeepIdle, <<NewIdle:32/native>>}
, {raw, 6, OptKeepInterval, <<NewInterval:32/native>>}
, {raw, 6, OptKeepCount, <<NewProbes:32/native>>}
]} = Transport:getopts(Socket, Opts),
?assertEqual(NewIdle, Idle + 1),
?assertEqual(NewInterval, Interval + 1),

4
test/emqx_ocsp_cache_SUITE.erl
View File

@ -65,7 +65,7 @@ end_per_group(_Group, _Config) ->
ok.
init_per_testcase(t_openssl_client, Config) ->
ct:timetrap(10_000),
ct:timetrap(30_000),
OriginalListeners = application:get_env(emqx, listeners),
DataDir = ?config(data_dir, Config),
IssuerPem = filename:join([DataDir, "ocsp-issuer.pem"]),
@ -99,6 +99,8 @@ init_per_testcase(t_openssl_client, Config) ->
OCSPOpts = [ {ocsp_stapling_enabled, true}
, {ocsp_responder_url, "http://127.0.0.1:9877"}
, {ocsp_issuer_pem, IssuerPem}
, {ocsp_refresh_http_timeout, 15_000}
, {ocsp_refresh_interval, 1_000}
],
Opts2 = emqx_misc:merge_opts(Opts1, [ {ocsp_options, OCSPOpts}
, {ssl_options, SSLOpts2}]),