Add INI-type configuration support

3 files changed, 284 insertions, 0 deletions

diff --git a/deps/rabbitmq_stream/priv/schema/rabbitmq_stream.schema b/deps/rabbitmq_stream/priv/schema/rabbitmq_stream.schema
new file mode 100644
index 0000000000..0dc66d5382
--- /dev/null
+++ b/deps/rabbitmq_stream/priv/schema/rabbitmq_stream.schema
@@ -0,0 +1,158 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2020 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+%% ==========================================================================
+%% ----------------------------------------------------------------------------
+%% RabbitMQ Stream Plugin
+%%
+%% See https://www.rabbitmq.com/stream.html for details
+%% ----------------------------------------------------------------------------
+
+%  {rabbitmq_stream,
+%   [%% Network Configuration - the format is generally the same as for the broker
+
+%% Listen only on localhost (ipv4 & ipv6) on a specific port.
+%% {tcp_listeners, [{"127.0.0.1", 5555},
+%%                  {"::1",       5555}]},
+
+{mapping, "stream.listeners.tcp", "rabbitmq_stream.tcp_listeners",[
+    {datatype, {enum, [none]}}
+]}.
+
+{mapping, "stream.listeners.tcp.$name", "rabbitmq_stream.tcp_listeners",[
+    {datatype, [integer, ip]}
+]}.
+
+{translation, "rabbitmq_stream.tcp_listeners",
+fun(Conf) ->
+    case cuttlefish:conf_get("stream.listeners.tcp", Conf, undefined) of
+        none -> [];
+        _ ->
+            Settings = cuttlefish_variable:filter_by_prefix("stream.listeners.tcp", Conf),
+            [ V || {_, V} <- Settings ]
+    end
+end}.
+
+{mapping, "stream.tcp_listen_options", "rabbitmq_stream.tcp_listen_options", [
+    {datatype, {enum, [none]}}]}.
+
+{translation, "rabbitmq_stream.tcp_listen_options",
+fun(Conf) ->
+    case cuttlefish:conf_get("stream.tcp_listen_options", Conf, undefined) of
+        none -> [];
+        _    -> cuttlefish:invalid("Invalid stream.tcp_listen_options")
+    end
+end}.
+
+{mapping, "stream.tcp_listen_options.backlog", "rabbitmq_stream.tcp_listen_options.backlog", [
+    {datatype, integer}
+]}.
+
+{mapping, "stream.tcp_listen_options.nodelay", "rabbitmq_stream.tcp_listen_options.nodelay", [
+    {datatype, {enum, [true, false]}}
+]}.
+
+{mapping, "stream.tcp_listen_options.buffer", "rabbitmq_stream.tcp_listen_options.buffer",
+    [{datatype, integer}]}.
+
+{mapping, "stream.tcp_listen_options.delay_send", "rabbitmq_stream.tcp_listen_options.delay_send",
+    [{datatype, {enum, [true, false]}}]}.
+
+{mapping, "stream.tcp_listen_options.dontroute", "rabbitmq_stream.tcp_listen_options.dontroute",
+    [{datatype, {enum, [true, false]}}]}.
+
+{mapping, "stream.tcp_listen_options.exit_on_close", "rabbitmq_stream.tcp_listen_options.exit_on_close",
+    [{datatype, {enum, [true, false]}}]}.
+
+{mapping, "stream.tcp_listen_options.fd", "rabbitmq_stream.tcp_listen_options.fd",
+    [{datatype, integer}]}.
+
+{mapping, "stream.tcp_listen_options.high_msgq_watermark", "rabbitmq_stream.tcp_listen_options.high_msgq_watermark",
+    [{datatype, integer}]}.
+
+{mapping, "stream.tcp_listen_options.high_watermark", "rabbitmq_stream.tcp_listen_options.high_watermark",
+    [{datatype, integer}]}.
+
+{mapping, "stream.tcp_listen_options.keepalive", "rabbitmq_stream.tcp_listen_options.keepalive",
+    [{datatype, {enum, [true, false]}}]}.
+
+{mapping, "stream.tcp_listen_options.low_msgq_watermark", "rabbitmq_stream.tcp_listen_options.low_msgq_watermark",
+    [{datatype, integer}]}.
+
+{mapping, "stream.tcp_listen_options.low_watermark", "rabbitmq_stream.tcp_listen_options.low_watermark",
+    [{datatype, integer}]}.
+
+{mapping, "stream.tcp_listen_options.port", "rabbitmq_stream.tcp_listen_options.port",
+    [{datatype, integer}, {validators, ["port"]}]}.
+
+{mapping, "stream.tcp_listen_options.priority", "rabbitmq_stream.tcp_listen_options.priority",
+    [{datatype, integer}]}.
+
+{mapping, "stream.tcp_listen_options.recbuf", "rabbitmq_stream.tcp_listen_options.recbuf",
+    [{datatype, integer}]}.
+
+{mapping, "stream.tcp_listen_options.send_timeout", "rabbitmq_stream.tcp_listen_options.send_timeout",
+    [{datatype, integer}]}.
+
+{mapping, "stream.tcp_listen_options.send_timeout_close", "rabbitmq_stream.tcp_listen_options.send_timeout_close",
+    [{datatype, {enum, [true, false]}}]}.
+
+{mapping, "stream.tcp_listen_options.sndbuf", "rabbitmq_stream.tcp_listen_options.sndbuf",
+    [{datatype, integer}]}.
+
+{mapping, "stream.tcp_listen_options.tos", "rabbitmq_stream.tcp_listen_options.tos",
+    [{datatype, integer}]}.
+
+{mapping, "stream.tcp_listen_options.linger.on", "rabbitmq_stream.tcp_listen_options.linger",
+    [{datatype, {enum, [true, false]}}]}.
+
+{mapping, "stream.tcp_listen_options.linger.timeout", "rabbitmq_stream.tcp_listen_options.linger",
+    [{datatype, integer}, {validators, ["non_negative_integer"]}]}.
+
+{translation, "rabbitmq_stream.tcp_listen_options.linger",
+fun(Conf) ->
+    LingerOn = cuttlefish:conf_get("stream.tcp_listen_options.linger.on", Conf, false),
+    LingerTimeout = cuttlefish:conf_get("stream.tcp_listen_options.linger.timeout", Conf, 0),
+    {LingerOn, LingerTimeout}
+end}.
+
+%% Number of Erlang processes that will accept connections for the TCP listener
+%%
+%% {num_tcp_acceptors, 10},
+
+{mapping, "stream.num_acceptors.tcp", "rabbitmq_stream.num_tcp_acceptors", [
+    {datatype, integer}
+]}.
+
+{mapping, "stream.initial_credits", "rabbitmq_stream.initial_credits", [
+    {datatype, integer}
+]}.
+
+{mapping, "stream.credits_required_for_unblocking", "rabbitmq_stream.credits_required_for_unblocking", [
+    {datatype, integer}
+]}.
+
+{mapping, "stream.frame_max", "rabbitmq_stream.frame_max", [
+    {datatype, integer}
+]}.
+
+{mapping, "stream.heartbeat", "rabbitmq_stream.heartbeat", [
+    {datatype, integer}
+]}.
+
+{mapping, "stream.advertised_host", "rabbitmq_stream.advertised_host", [
+    {datatype, string}
+]}.
+
+{translation, "rabbitmq_stream.advertised_host",
+fun(Conf) ->
+    list_to_binary(cuttlefish:conf_get("stream.advertised_host", Conf))
+end}.
+
+{mapping, "stream.advertised_port", "rabbitmq_stream.advertised_port", [
+    {datatype, integer}
+]}.
\ No newline at end of file
diff --git a/deps/rabbitmq_stream/test/config_schema_SUITE.erl b/deps/rabbitmq_stream/test/config_schema_SUITE.erl
new file mode 100644
index 0000000000..a298811541
--- /dev/null
+++ b/deps/rabbitmq_stream/test/config_schema_SUITE.erl
@@ -0,0 +1,53 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2020 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(config_schema_SUITE).
+
+-compile(export_all).
+
+all() ->
+  [
+    run_snippets
+  ].
+
+%% -------------------------------------------------------------------
+%% Testsuite setup/teardown.
+%% -------------------------------------------------------------------
+
+init_per_suite(Config) ->
+  rabbit_ct_helpers:log_environment(),
+  Config1 = rabbit_ct_helpers:run_setup_steps(Config),
+  rabbit_ct_config_schema:init_schemas(rabbitmq_stream, Config1).
+
+end_per_suite(Config) ->
+  rabbit_ct_helpers:run_teardown_steps(Config).
+
+init_per_testcase(Testcase, Config) ->
+  rabbit_ct_helpers:testcase_started(Config, Testcase),
+  Config1 = rabbit_ct_helpers:set_config(Config, [
+    {rmq_nodename_suffix, Testcase}
+  ]),
+  rabbit_ct_helpers:run_steps(Config1,
+    rabbit_ct_broker_helpers:setup_steps() ++
+    rabbit_ct_client_helpers:setup_steps()).
+
+end_per_testcase(Testcase, Config) ->
+  Config1 = rabbit_ct_helpers:run_steps(Config,
+    rabbit_ct_client_helpers:teardown_steps() ++
+    rabbit_ct_broker_helpers:teardown_steps()),
+  rabbit_ct_helpers:testcase_finished(Config1, Testcase).
+
+%% -------------------------------------------------------------------
+%% Testcases.
+%% -------------------------------------------------------------------
+
+run_snippets(Config) ->
+  ok = rabbit_ct_broker_helpers:rpc(Config, 0,
+    ?MODULE, run_snippets1, [Config]).
+
+run_snippets1(Config) ->
+  rabbit_ct_config_schema:run_snippets(Config).
diff --git a/deps/rabbitmq_stream/test/config_schema_SUITE_data/rabbitmq_stream.snippets b/deps/rabbitmq_stream/test/config_schema_SUITE_data/rabbitmq_stream.snippets
new file mode 100644
index 0000000000..8f60ef9710
--- /dev/null
+++ b/deps/rabbitmq_stream/test/config_schema_SUITE_data/rabbitmq_stream.snippets
@@ -0,0 +1,73 @@
+[{listener_port,
+  "stream.listeners.tcp.1 = 12345",
+  [{rabbitmq_stream,[{tcp_listeners,[12345]}]}],
+  [rabbitmq_stream]},
+ {listeners_ip,
+  "stream.listeners.tcp.1 = 127.0.0.1:5555
+   stream.listeners.tcp.2 = ::1:5555",
+  [{rabbitmq_stream,[{tcp_listeners,[{"127.0.0.1",5555},{"::1",5555}]}]}],
+  [rabbitmq_stream]},
+
+ {listener_tcp_options,
+  "stream.listeners.tcp.1 = 127.0.0.1:5555
+   stream.listeners.tcp.2 = ::1:5555
+
+   stream.tcp_listen_options.backlog = 2048
+   stream.tcp_listen_options.recbuf = 8192
+   stream.tcp_listen_options.sndbuf = 8192
+
+   stream.tcp_listen_options.keepalive = true
+   stream.tcp_listen_options.nodelay   = true
+
+   stream.tcp_listen_options.exit_on_close = true
+
+   stream.tcp_listen_options.send_timeout = 120
+",
+  [{rabbitmq_stream,[
+                    {tcp_listeners,[
+                                    {"127.0.0.1",5555},
+                                    {"::1",5555}
+                                   ]}
+                    , {tcp_listen_options, [
+                                            {backlog, 2048},
+                                            {exit_on_close, true},
+
+                                            {recbuf, 8192},
+                                            {sndbuf, 8192},
+
+                                            {send_timeout, 120},
+
+                                            {keepalive, true},
+                                            {nodelay,   true}
+                                           ]}
+                   ]}],
+  [rabbitmq_stream]},
+ {defaults,
+  "stream.frame_max = 1048576
+   stream.heartbeat = 60
+   stream.initial_credits = 50000
+   stream.credits_required_for_unblocking = 12500",
+  [{rabbitmq_stream,[{initial_credits, 50000},
+                     	{credits_required_for_unblocking, 12500},
+                     	{frame_max, 1048576},
+                     	{heartbeat, 60}]}],
+  [rabbitmq_stream]},
+  {advertised_host_port,
+      "stream.advertised_host = some-host
+       stream.advertised_port = 5556",
+      [{rabbitmq_stream,[{advertised_host, <<"some-host">>},
+                         	{advertised_port, 5556}]}],
+      [rabbitmq_stream]},
+  {credits,
+    "stream.frame_max = 2097152
+        stream.heartbeat = 120",
+    [{rabbitmq_stream,[{frame_max, 2097152},
+                       	{heartbeat, 120}]}],
+    [rabbitmq_stream]},
+  {protocol,
+      "stream.initial_credits = 100000
+          stream.credits_required_for_unblocking = 25000",
+      [{rabbitmq_stream,[{initial_credits, 100000},
+                         	{credits_required_for_unblocking, 25000}]}],
+      [rabbitmq_stream]}
+].
\ No newline at end of file