Merge branch 'master' into management-only-api

Author: michaelklishin

erlang.mk

@@ -65,15 +65,15 @@
 
 -define(amqqueue_has_valid_pid(Q),
         ((?is_amqqueue_v2(Q) andalso
-         is_pid(?amqqueue_v2_field_pid(Q))) orelse
-        (?is_amqqueue_v1(Q) andalso
-         is_pid(?amqqueue_v1_field_pid(Q))))).
+          is_pid(?amqqueue_v2_field_pid(Q))) orelse
+         (?is_amqqueue_v1(Q) andalso
+          is_pid(?amqqueue_v1_field_pid(Q))))).
 
 -define(amqqueue_pid_runs_on_local_node(Q),
         ((?is_amqqueue_v2(Q) andalso
-         node(?amqqueue_v2_field_pid(Q)) =:= node()) orelse
-        (?is_amqqueue_v1(Q) andalso
-         node(?amqqueue_v1_field_pid(Q)) =:= node()))).
+          node(?amqqueue_v2_field_pid(Q)) =:= node()) orelse
+         (?is_amqqueue_v1(Q) andalso
+          node(?amqqueue_v1_field_pid(Q)) =:= node()))).
 
 -define(amqqueue_pid_equals(Q, Pid),
         ((?is_amqqueue_v2(Q) andalso
@@ -108,9 +108,9 @@
 
 -define(amqqueue_vhost_equals(Q, VHost),
         ((?is_amqqueue_v2(Q) andalso
-         ?amqqueue_v2_vhost(Q) =:= VHost) orelse
-        (?is_amqqueue_v1(Q) andalso
-         ?amqqueue_v1_vhost(Q) =:= VHost))).
+          ?amqqueue_v2_vhost(Q) =:= VHost) orelse
+         (?is_amqqueue_v1(Q) andalso
+          ?amqqueue_v1_vhost(Q) =:= VHost))).
 
 -ifdef(DEBUG_QUORUM_QUEUE_FF).
 -define(enable_quorum_queue_if_debug,

include/amqqueue.hrl

@@ -391,6 +391,13 @@ end}.
     {datatype, integer}
 ]}.
 
+%% Default worker process pool size. Used to limit maximum concurrency rate
+%% of certain operations, e.g. queue initialisation and recovery on node boot.
+
+{mapping, "default_worker_pool_size", "rabbit.default_worker_pool_size", [
+    {datatype, integer}, {validators, ["non_negative_integer"]}
+]}.
+
 %% Password hashing implementation. Will only affect newly
 %% created users. To recalculate hash for an existing user
 %% it's necessary to update her password.

priv/schema/rabbit.schema

@@ -184,6 +184,15 @@ RABBITMQ_COMPONENTS = amqp_client \
 		      rabbitmq_web_stomp_examples \
 		      rabbitmq_website
 
+# Erlang.mk does not rebuild dependencies by default, once they were
+# compiled once, except for those listed in the `$(FORCE_REBUILD)`
+# variable.
+#
+# We want all RabbitMQ components to always be rebuilt: this eases
+# the work on several components at the same time.
+
+FORCE_REBUILD = $(RABBITMQ_COMPONENTS)
+
 # Several components have a custom erlang.mk/build.config, mainly
 # to disable eunit. Therefore, we can't use the top-level project's
 # erlang.mk copy.

rabbitmq-components.mk

@@ -194,6 +194,7 @@ RABBITMQ_DIST_PORT=$RABBITMQ_DIST_PORT \
     -noinput \
     -hidden \
     -s rabbit_prelaunch \
+    ${RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS} \
     ${RABBITMQ_NAME_TYPE} ${RABBITMQ_PRELAUNCH_NODENAME} \
     -conf_advanced "${RABBITMQ_ADVANCED_CONFIG_FILE}" \
     -rabbit feature_flags_file "\"$RABBITMQ_FEATURE_FLAGS_FILE\"" \

scripts/rabbitmq-server

@@ -19,7 +19,9 @@
 -include_lib("rabbit_common/include/rabbit.hrl").
 -include("amqqueue.hrl").
 
--export([new/9,
+-export([new/8,
+         new/9,
+         new_with_version/9,
          new_with_version/10,
          fields/0,
          fields/1,
@@ -177,6 +179,40 @@
               amqqueue_v2_pattern/0,
               ra_server_id/0]).
 
+-spec new(rabbit_amqqueue:name(),
+          pid() | ra_server_id() | none,
+          boolean(),
+          boolean(),
+          pid() | none,
+          rabbit_framing:amqp_table(),
+          rabbit_types:vhost() | undefined,
+          map()) -> amqqueue().
+
+new(#resource{kind = queue} = Name,
+    Pid,
+    Durable,
+    AutoDelete,
+    Owner,
+    Args,
+    VHost,
+    Options)
+  when (is_pid(Pid) orelse is_tuple(Pid) orelse Pid =:= none) andalso
+       is_boolean(Durable) andalso
+       is_boolean(AutoDelete) andalso
+       (is_pid(Owner) orelse Owner =:= none) andalso
+       is_list(Args) andalso
+       (is_binary(VHost) orelse VHost =:= undefined) andalso
+       is_map(Options) ->
+    new(Name,
+        Pid,
+        Durable,
+        AutoDelete,
+        Owner,
+        Args,
+        VHost,
+        Options,
+        ?amqqueue_v1_type).
+
 -spec new(rabbit_amqqueue:name(),
           pid() | ra_server_id() | none,
           boolean(),
@@ -229,6 +265,44 @@ new(#resource{kind = queue} = Name,
               Options)
     end.
 
+-spec new_with_version
+(amqqueue_v1 | amqqueue_v2,
+ rabbit_amqqueue:name(),
+ pid() | ra_server_id() | none,
+ boolean(),
+ boolean(),
+ pid() | none,
+ rabbit_framing:amqp_table(),
+ rabbit_types:vhost() | undefined,
+ map()) -> amqqueue().
+
+new_with_version(RecordVersion,
+                 #resource{kind = queue} = Name,
+                 Pid,
+                 Durable,
+                 AutoDelete,
+                 Owner,
+                 Args,
+                 VHost,
+                 Options)
+  when (is_pid(Pid) orelse is_tuple(Pid) orelse Pid =:= none) andalso
+       is_boolean(Durable) andalso
+       is_boolean(AutoDelete) andalso
+       (is_pid(Owner) orelse Owner =:= none) andalso
+       is_list(Args) andalso
+       (is_binary(VHost) orelse VHost =:= undefined) andalso
+       is_map(Options) ->
+    new_with_version(RecordVersion,
+                     Name,
+                     Pid,
+                     Durable,
+                     AutoDelete,
+                     Owner,
+                     Args,
+                     VHost,
+                     Options,
+                     ?amqqueue_v1_type).
+
 -spec new_with_version
 (amqqueue_v1 | amqqueue_v2,
  rabbit_amqqueue:name(),
@@ -359,6 +433,8 @@ get_exclusive_owner(#amqqueue{exclusive_owner = Owner}) ->
 get_exclusive_owner(Queue) ->
     amqqueue_v1:get_exclusive_owner(Queue).
 
+% gm_pids
+
 -spec get_gm_pids(amqqueue()) -> [{pid(), pid()} | pid()] | none.
 
 get_gm_pids(#amqqueue{gm_pids = GMPids}) ->
@@ -421,7 +497,7 @@ get_pid(#amqqueue{pid = Pid}) -> Pid;
 get_pid(Queue)                -> amqqueue_v1:get_pid(Queue).
 
 -spec set_pid
-(amqqueue_v2(),  pid() | ra_server_id() | none) -> amqqueue_v2();
+(amqqueue_v2(), pid() | ra_server_id() | none) -> amqqueue_v2();
 (amqqueue_v1:amqqueue_v1(), pid() | none) -> amqqueue_v1:amqqueue_v1().
 
 set_pid(#amqqueue{} = Queue, Pid) ->
@@ -509,7 +585,8 @@ set_slave_pids(Queue, SlavePids) ->
 
 -spec get_slave_pids_pending_shutdown(amqqueue()) -> [pid()].
 
-get_slave_pids_pending_shutdown(#amqqueue{slave_pids_pending_shutdown = Slaves}) ->
+get_slave_pids_pending_shutdown(
+  #amqqueue{slave_pids_pending_shutdown = Slaves}) ->
     Slaves;
 get_slave_pids_pending_shutdown(Queue) ->
     amqqueue_v1:get_slave_pids_pending_shutdown(Queue).