Merge pull request #8148 from zhongwencool/listener-options

feat: make tcp/ssl options more straightforward

apps/emqx/etc/emqx.conf

@@ -1,39 +1,39 @@
 listeners.tcp.default {
-  bind: "0.0.0.0:1883"
-  max_connections: 1024000
+  bind = "0.0.0.0:1883"
+  max_connections = 1024000
 }
 
 listeners.ssl.default {
-  bind: "0.0.0.0:8883"
-  max_connections: 512000
-  ssl {
-    keyfile: "{{ platform_etc_dir }}/certs/key.pem"
-    certfile: "{{ platform_etc_dir }}/certs/cert.pem"
-    cacertfile: "{{ platform_etc_dir }}/certs/cacert.pem"
+  bind = "0.0.0.0:8883"
+  max_connections = 512000
+  ssl_options {
+    keyfile = "{{ platform_etc_dir }}/certs/key.pem"
+    certfile = "{{ platform_etc_dir }}/certs/cert.pem"
+    cacertfile = "{{ platform_etc_dir }}/certs/cacert.pem"
   }
 }
 
 listeners.ws.default {
-  bind: "0.0.0.0:8083"
-  max_connections: 1024000
-  websocket.mqtt_path: "/mqtt"
+  bind = "0.0.0.0:8083"
+  max_connections = 1024000
+  websocket.mqtt_path = "/mqtt"
 }
 
 listeners.wss.default {
-  bind: "0.0.0.0:8084"
-  max_connections: 512000
-  websocket.mqtt_path: "/mqtt"
-  ssl {
-    keyfile: "{{ platform_etc_dir }}/certs/key.pem"
-    certfile: "{{ platform_etc_dir }}/certs/cert.pem"
-    cacertfile: "{{ platform_etc_dir }}/certs/cacert.pem"
+  bind = "0.0.0.0:8084"
+  max_connections = 512000
+  websocket.mqtt_path = "/mqtt"
+  ssl_options {
+    keyfile = "{{ platform_etc_dir }}/certs/key.pem"
+    certfile = "{{ platform_etc_dir }}/certs/cert.pem"
+    cacertfile = "{{ platform_etc_dir }}/certs/cacert.pem"
   }
 }
 
 # listeners.quic.default {
-#  enabled: false
-#  bind: "0.0.0.0:14567"
-#  max_connections: 1024000
-#  keyfile: "{{ platform_etc_dir }}/certs/key.pem"
-#  certfile: "{{ platform_etc_dir }}/certs/cert.pem"
+#  enabled = false
+#  bind = "0.0.0.0:14567"
+#  max_connections = 1024000
+#  keyfile = "{{ platform_etc_dir }}/certs/key.pem"
+#  certfile = "{{ platform_etc_dir }}/certs/cert.pem"
 #}

apps/emqx/src/emqx_connection.erl

@@ -1176,5 +1176,7 @@ get_state(Pid) ->
         )
     ).
 
-get_active_n(quic, _Listener) -> ?ACTIVE_N;
-get_active_n(Type, Listener) -> emqx_config:get_listener_conf(Type, Listener, [tcp, active_n]).
+get_active_n(quic, _Listener) ->
+    ?ACTIVE_N;
+get_active_n(Type, Listener) ->
+    emqx_config:get_listener_conf(Type, Listener, [tcp_options, active_n]).

apps/emqx/src/emqx_listeners.erl

@@ -406,7 +406,7 @@ esockd_opts(Type, Opts0) ->
 
 ws_opts(Type, ListenerName, Opts) ->
     WsPaths = [
-        {maps:get(mqtt_path, Opts, "/mqtt"), emqx_ws_connection, #{
+        {emqx_map_lib:deep_get([websocket, mqtt_path], Opts, "/mqtt"), emqx_ws_connection, #{
             zone => zone(Opts),
             listener => {Type, ListenerName},
             limiter => limiter(Opts)
@@ -497,7 +497,7 @@ limiter(Opts) ->
 ssl_opts(Opts) ->
     maps:to_list(
         emqx_tls_lib:drop_tls13_for_old_otp(
-            maps:get(ssl, Opts, #{})
+            maps:get(ssl_options, Opts, #{})
         )
     ).
 
@@ -505,7 +505,7 @@ tcp_opts(Opts) ->
     maps:to_list(
         maps:without(
             [active_n],
-            maps:get(tcp, Opts, #{})
+            maps:get(tcp_options, Opts, #{})
         )
     ).
 
@@ -557,18 +557,18 @@ certs_dir(Type, Name) ->
     iolist_to_binary(filename:join(["listeners", Type, Name])).
 
 convert_certs(CertsDir, Conf) ->
-    case emqx_tls_lib:ensure_ssl_files(CertsDir, maps:get(<<"ssl">>, Conf, undefined)) of
+    case emqx_tls_lib:ensure_ssl_files(CertsDir, maps:get(<<"ssl_options">>, Conf, undefined)) of
         {ok, undefined} ->
             Conf;
         {ok, SSL} ->
-            Conf#{<<"ssl">> => SSL};
+            Conf#{<<"ssl_options">> => SSL};
         {error, Reason} ->
             ?SLOG(error, Reason#{msg => "bad_ssl_config"}),
             throw({bad_ssl_config, Reason})
     end.
 
 clear_certs(CertsDir, Conf) ->
-    OldSSL = maps:get(<<"ssl">>, Conf, undefined),
+    OldSSL = maps:get(<<"ssl_options">>, Conf, undefined),
     emqx_tls_lib:delete_ssl_files(CertsDir, undefined, OldSSL).
 
 filter_stacktrace({Reason, _Stacktrace}) -> Reason;

apps/emqx/src/emqx_schema.erl

@@ -787,57 +787,61 @@ fields("listeners") ->
             )}
     ];
 fields("mqtt_tcp_listener") ->
-    [
-        {"tcp",
-            sc(
-                ref("tcp_opts"),
-                #{}
-            )}
-    ] ++ mqtt_listener(1883);
+    mqtt_listener(1883) ++
+        [
+            {"tcp_options",
+                sc(
+                    ref("tcp_opts"),
+                    #{}
+                )}
+        ];
 fields("mqtt_ssl_listener") ->
-    [
-        {"tcp",
-            sc(
-                ref("tcp_opts"),
-                #{}
-            )},
-        {"ssl",
-            sc(
-                ref("listener_ssl_opts"),
-                #{}
-            )}
-    ] ++ mqtt_listener(8883);
+    mqtt_listener(8883) ++
+        [
+            {"tcp_options",
+                sc(
+                    ref("tcp_opts"),
+                    #{}
+                )},
+            {"ssl_options",
+                sc(
+                    ref("listener_ssl_opts"),
+                    #{}
+                )}
+        ];
 fields("mqtt_ws_listener") ->
-    [
-        {"tcp",
-            sc(
-                ref("tcp_opts"),
-                #{}
-            )},
-        {"websocket",
-            sc(
-                ref("ws_opts"),
-                #{}
-            )}
-    ] ++ mqtt_listener(8083);
+    mqtt_listener(8083) ++
+        [
+            {"tcp_options",
+                sc(
+                    ref("tcp_opts"),
+                    #{}
+                )},
+            {"websocket",
+                sc(
+                    ref("ws_opts"),
+                    #{}
+                )}
+        ];
 fields("mqtt_wss_listener") ->
-    [
-        {"tcp",
-            sc(
-                ref("tcp_opts"),
-                #{}
-            )},
-        {"ssl",
-            sc(
-                ref("listener_wss_opts"),
-                #{}
-            )},
-        {"websocket",
-            sc(
-                ref("ws_opts"),
-                #{}
-            )}
-    ] ++ mqtt_listener(8084);
+    mqtt_listener(8084) ++
+        [
+            {"tcp_options",
+                sc(
+                    ref("tcp_opts"),
+                    #{}
+                )},
+            {"ssl_options",
+                sc(
+                    ref("listener_wss_opts"),
+                    #{}
+                )},
+            {"websocket",
+                sc(
+                    ref("ws_opts"),
+                    #{}
+                )}
+        ];
 fields("mqtt_quic_listener") ->
     [
         {"enabled",

apps/emqx/src/emqx_ws_connection.erl

@@ -1046,4 +1046,4 @@ get_ws_opts(Type, Listener, Key) ->
     emqx_config:get_listener_conf(Type, Listener, [websocket, Key]).
 
 get_active_n(Type, Listener) ->
-    emqx_config:get_listener_conf(Type, Listener, [tcp, active_n]).
+    emqx_config:get_listener_conf(Type, Listener, [tcp_options, active_n]).

apps/emqx/test/emqx_channel_SUITE.erl

@@ -106,7 +106,7 @@ listener_mqtt_tcp_conf() ->
         mountpoint => <<>>,
         proxy_protocol => false,
         proxy_protocol_timeout => 3000,
-        tcp => #{
+        tcp_options => #{
             active_n => 100,
             backlog => 1024,
             buffer => 4096,
@@ -128,7 +128,7 @@ listener_mqtt_ws_conf() ->
         mountpoint => <<>>,
         proxy_protocol => false,
         proxy_protocol_timeout => 3000,
-        tcp =>
+        tcp_options =>
             #{
                 active_n => 100,
                 backlog => 1024,

apps/emqx/test/emqx_client_SUITE.erl

@@ -78,7 +78,7 @@ groups() ->
 init_per_suite(Config) ->
     emqx_common_test_helpers:boot_modules(all),
     emqx_common_test_helpers:start_apps([]),
-    emqx_config:put_listener_conf(ssl, default, [ssl, verify], verify_peer),
+    emqx_config:put_listener_conf(ssl, default, [ssl_options, verify], verify_peer),
     emqx_listeners:restart_listener('ssl:default'),
     Config.
 

apps/emqx/test/emqx_connection_SUITE.erl

@@ -256,9 +256,9 @@ t_handle_msg_deliver(_) ->
 
 t_handle_msg_inet_reply(_) ->
     ok = meck:expect(emqx_pd, get_counter, fun(_) -> 10 end),
-    emqx_config:put_listener_conf(tcp, default, [tcp, active_n], 0),
+    emqx_config:put_listener_conf(tcp, default, [tcp_options, active_n], 0),
     ?assertMatch({ok, _St}, handle_msg({inet_reply, for_testing, ok}, st())),
-    emqx_config:put_listener_conf(tcp, default, [tcp, active_n], 100),
+    emqx_config:put_listener_conf(tcp, default, [tcp_options, active_n], 100),
     ?assertEqual(ok, handle_msg({inet_reply, for_testing, ok}, st())),
     ?assertMatch(
         {stop, {shutdown, for_testing}, _St},

apps/emqx/test/emqx_listeners_SUITE.erl

@@ -78,7 +78,7 @@ init_per_testcase(t_wss_conn, Config) ->
             listener_test => #{
                 bind => {{127, 0, 0, 1}, 9998},
                 limiter => #{},
-                ssl => #{
+                ssl_options => #{
                     cacertfile => ?CERTS_PATH("cacert.pem"),
                     certfile => ?CERTS_PATH("cert.pem"),
                     keyfile => ?CERTS_PATH("key.pem")

apps/emqx/test/emqx_mqtt_protocol_v5_SUITE.erl

@@ -70,6 +70,9 @@ init_per_group(_, Config) ->
     emqx_common_test_helpers:stop_apps([]),
     Config.
 
+end_per_group(quic, _Config) ->
+    emqx_config:put([listeners, quic], #{}),
+    ok;
 end_per_group(_Group, _Config) ->
     ok.
 

apps/emqx/test/emqx_takeover_SUITE.erl

@@ -33,11 +33,12 @@
 all() -> emqx_common_test_helpers:all(?MODULE).
 
 init_per_suite(Config) ->
+    emqx_channel_SUITE:set_test_listener_confs(),
     ?check_trace(
         ?wait_async_action(
             emqx_common_test_helpers:start_apps([]),
             #{?snk_kind := listener_started, bind := 1883},
-            timer:seconds(5)
+            timer:seconds(10)
         ),
         fun(Trace) ->
             %% more than one listener

apps/emqx_management/src/emqx_mgmt_api_listeners.erl

@@ -685,7 +685,7 @@ tcp_schema_example() ->
         proxy_protocol => false,
         proxy_protocol_timeout => <<"3s">>,
         running => true,
-        tcp => #{
+        tcp_options => #{
             active_n => 100,
             backlog => 1024,
             buffer => <<"4KB">>,

apps/emqx_management/test/emqx_mgmt_api_listeners_SUITE.erl

@@ -96,7 +96,7 @@ crud_listeners_by_id(ListenerId, NewListenerId, MinListenerId, BadId, Type) ->
     MinConf =
         case OriginListener of
             #{
-                <<"ssl">> :=
+                <<"ssl_options">> :=
                     #{
                         <<"cacertfile">> := CaCertFile,
                         <<"certfile">> := CertFile,
@@ -107,7 +107,7 @@ crud_listeners_by_id(ListenerId, NewListenerId, MinListenerId, BadId, Type) ->
                     <<"id">> => MinListenerId,
                     <<"bind">> => <<"0.0.0.0:3883">>,
                     <<"type">> => Type,
-                    <<"ssl">> => #{
+                    <<"ssl_options">> => #{
                         <<"cacertfile">> => CaCertFile,
                         <<"certfile">> => CertFile,
                         <<"keyfile">> => KeyFile

apps/emqx_modules/test/emqx_delayed_SUITE.erl

@@ -30,11 +30,18 @@
 %%--------------------------------------------------------------------
 %% Setups
 %%--------------------------------------------------------------------
+-define(BASE_CONF, #{
+    <<"dealyed">> => <<"true">>,
+    <<"max_delayed_messages">> => <<"0">>
+}).
 
 all() ->
     emqx_common_test_helpers:all(?MODULE).
 
 init_per_suite(Config) ->
+    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, ?BASE_CONF, #{
+        raw_with_default => true
+    }),
     emqx_common_test_helpers:start_apps([emqx_conf, emqx_modules]),
     Config.
 

apps/emqx_modules/test/emqx_delayed_api_SUITE.erl

@@ -32,7 +32,7 @@ all() ->
     emqx_common_test_helpers:all(?MODULE).
 
 init_per_suite(Config) ->
-    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, jsx:encode(?BASE_CONF), #{
+    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, ?BASE_CONF, #{
         raw_with_default => true
     }),
 

apps/emqx_modules/test/emqx_rewrite_SUITE.erl

@@ -160,7 +160,7 @@ t_rewrite_re_error(_Config) ->
     ok.
 
 t_list(_Config) ->
-    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, jsx:encode(?REWRITE), #{
+    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, ?REWRITE, #{
         raw_with_default => true
     }),
     Expect = maps:get(<<"rewrite">>, ?REWRITE),
@@ -168,7 +168,7 @@ t_list(_Config) ->
     ok.
 
 t_update(_Config) ->
-    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, jsx:encode(?REWRITE), #{
+    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, ?REWRITE, #{
         raw_with_default => true
     }),
     Init = emqx_rewrite:list(),
@@ -186,7 +186,7 @@ t_update(_Config) ->
     ok.
 
 t_update_disable(_Config) ->
-    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, jsx:encode(?REWRITE), #{
+    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, ?REWRITE, #{
         raw_with_default => true
     }),
     ?assertEqual(ok, emqx_rewrite:update([])),
@@ -203,7 +203,7 @@ t_update_disable(_Config) ->
     ok.
 
 t_update_re_failed(_Config) ->
-    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, jsx:encode(?REWRITE), #{
+    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, ?REWRITE, #{
         raw_with_default => true
     }),
     Re = <<"*^test/*">>,
@@ -260,7 +260,7 @@ receive_publish(Timeout) ->
     end.
 
 init() ->
-    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, jsx:encode(?REWRITE), #{
+    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, ?REWRITE, #{
         raw_with_default => true
     }),
     ok = emqx_rewrite:enable(),

apps/emqx_modules/test/emqx_rewrite_api_SUITE.erl

@@ -33,7 +33,7 @@ init_per_testcase(_, Config) ->
     Config.
 
 init_per_suite(Config) ->
-    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, jsx:encode(?BASE_CONF), #{
+    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, ?BASE_CONF, #{
         raw_with_default => true
     }),
 

apps/emqx_modules/test/emqx_telemetry_SUITE.erl

@@ -41,7 +41,7 @@ init_per_suite(Config) ->
             emqx_common_test_helpers:deps_path(emqx_authz, "etc/acl.conf")
         end
     ),
-    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, jsx:encode(?BASE_CONF), #{
+    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, ?BASE_CONF, #{
         raw_with_default => true
     }),
     emqx_common_test_helpers:start_apps(
@@ -176,7 +176,7 @@ init_per_testcase(t_uuid_restored_from_file, Config) ->
     %% clear the UUIDs in the DB
     {atomic, ok} = mria:clear_table(emqx_telemetry),
     emqx_common_test_helpers:stop_apps([emqx_conf, emqx_authn, emqx_authz, emqx_modules]),
-    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, jsx:encode(?BASE_CONF), #{
+    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, ?BASE_CONF, #{
         raw_with_default => true
     }),
     emqx_common_test_helpers:start_apps(
@@ -332,7 +332,7 @@ t_uuid_saved_to_file(_Config) ->
     %% clear the UUIDs in the DB
     {atomic, ok} = mria:clear_table(emqx_telemetry),
     emqx_common_test_helpers:stop_apps([emqx_conf, emqx_authn, emqx_authz, emqx_modules]),
-    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, jsx:encode(?BASE_CONF), #{
+    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, ?BASE_CONF, #{
         raw_with_default => true
     }),
     emqx_common_test_helpers:start_apps(
@@ -861,7 +861,7 @@ setup_slave(Node) ->
         Node,
         emqx_common_test_helpers,
         load_config,
-        [emqx_modules_schema, jsx:encode(?BASE_CONF), #{raw_with_default => true}]
+        [emqx_modules_schema, ?BASE_CONF, #{raw_with_default => true}]
     ),
     ok = rpc:call(
         Node,

apps/emqx_modules/test/emqx_telemetry_api_SUITE.erl

@@ -29,7 +29,7 @@ all() ->
     emqx_common_test_helpers:all(?MODULE).
 
 init_per_suite(Config) ->
-    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, jsx:encode(?BASE_CONF), #{
+    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, ?BASE_CONF, #{
         raw_with_default => true
     }),
 

apps/emqx_modules/test/emqx_topic_metrics_SUITE.erl

@@ -28,7 +28,7 @@ all() -> emqx_common_test_helpers:all(?MODULE).
 
 init_per_suite(Config) ->
     emqx_common_test_helpers:boot_modules(all),
-    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, jsx:encode(?TOPIC), #{
+    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, ?TOPIC, #{
         raw_with_default => true
     }),
     emqx_common_test_helpers:start_apps([emqx_conf, emqx_modules]),

apps/emqx_modules/test/emqx_topic_metrics_api_SUITE.erl

@@ -40,7 +40,7 @@ init_per_testcase(_, Config) ->
     Config.
 
 init_per_suite(Config) ->
-    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, jsx:encode(?BASE_CONF), #{
+    ok = emqx_common_test_helpers:load_config(emqx_modules_schema, ?BASE_CONF, #{
         raw_with_default => true
     }),
 

apps/emqx_psk/test/emqx_psk_SUITE.erl

@@ -84,7 +84,7 @@ t_psk_lookup(_) ->
         reuseaddr => true,
         user_lookup_fun => {fun emqx_tls_psk:lookup/3, undefined}
     },
-    emqx_config:put([listeners, ssl, default, ssl], ServerTLSOpts),
+    emqx_config:put([listeners, ssl, default, ssl_options], ServerTLSOpts),
     emqx_listeners:restart_listener('ssl:default'),
 
     {ok, Socket} = ssl:connect("127.0.0.1", 8883, maps:to_list(ClientTLSOpts)),