Hard cap for maximum priorities

src/rabbit_amqqueue.erl

@@ -575,7 +575,7 @@ declare_args() ->
      {<<"x-dead-letter-routing-key">>, fun check_dlxrk_arg/2},
      {<<"x-max-length">>,              fun check_non_neg_int_arg/2},
      {<<"x-max-length-bytes">>,        fun check_non_neg_int_arg/2},
-     {<<"x-max-priority">>,            fun check_non_neg_int_arg/2},
+     {<<"x-max-priority">>,            fun check_max_priority_arg/2},
      {<<"x-overflow">>,                fun check_overflow/2},
      {<<"x-queue-mode">>,              fun check_queue_mode/2}].
 
@@ -611,6 +611,13 @@ check_message_ttl_arg({Type, Val}, Args) ->
         Error -> Error
     end.
 
+check_max_priority_arg({Type, Val}, Args) ->
+    case check_non_neg_int_arg({Type, Val}, Args) of
+        ok when Val =< ?MAX_SUPPORTED_PRIORITY -> ok;
+        ok                                     -> {error, {max_value_exceeded, Val}};
+        Error                                  -> Error
+    end.
+
 %% Note that the validity of x-dead-letter-exchange is already verified
 %% by rabbit_channel's queue.declare handler.
 check_dlxname_arg({longstr, _}, _) -> ok;

src/rabbit_ctl_usage.erl

@@ -0,0 +1,135 @@
+%% Generated, do not edit!
+-module(rabbit_ctl_usage).
+-export([usage/0]).
+usage() -> "Usage:
+rabbitmqctl [-n <node>] [-t <timeout>] [-q] <command> [<command options>] 
+
+Options:
+    -n node
+    -q
+    -t timeout
+
+Default node is \"rabbit@server\", where server is the local host. On a host 
+named \"server.example.com\", the node name of the RabbitMQ Erlang node will 
+usually be rabbit@server (unless RABBITMQ_NODENAME has been set to some 
+non-default value at broker startup time). The output of hostname -s is usually 
+the correct suffix to use after the \"@\" sign. See rabbitmq-server(1) for 
+details of configuring the RabbitMQ broker.
+
+Quiet output mode is selected with the \"-q\" flag. Informational messages are 
+suppressed when quiet mode is in effect.
+
+Operation timeout in seconds. Only applicable to \"list\" commands. Default is 
+\"infinity\".
+
+Commands:
+    stop [<pid_file>]
+    shutdown
+    stop_app
+    start_app
+    wait <pid_file>
+    reset
+    force_reset
+    rotate_logs <suffix>
+    hipe_compile <directory>
+
+    join_cluster <clusternode> [--ram]
+    cluster_status
+    change_cluster_node_type disc | ram
+    forget_cluster_node [--offline]
+    rename_cluster_node oldnode1 newnode1 [oldnode2] [newnode2 ...]
+    update_cluster_nodes clusternode
+    force_boot
+    sync_queue [-p <vhost>] queue
+    cancel_sync_queue [-p <vhost>] queue
+    purge_queue [-p <vhost>] queue
+    set_cluster_name name
+
+    add_user <username> <password>
+    delete_user <username>
+    change_password <username> <newpassword>
+    clear_password <username>
+    authenticate_user <username> <password>
+    set_user_tags <username> <tag> ...
+    list_users
+
+    add_vhost <vhost>
+    delete_vhost <vhost>
+    list_vhosts [<vhostinfoitem> ...]
+    set_permissions [-p <vhost>] <user> <conf> <write> <read>
+    clear_permissions [-p <vhost>] <username>
+    list_permissions [-p <vhost>]
+    list_user_permissions <username>
+
+    set_parameter [-p <vhost>] <component_name> <name> <value>
+    clear_parameter [-p <vhost>] <component_name> <key>
+    list_parameters [-p <vhost>]
+    set_global_parameter <name> <value>
+    clear_global_parameter <name>
+    list_global_parameters
+
+    set_policy [-p <vhost>] [--priority <priority>] [--apply-to <apply-to>] 
+<name> <pattern>  <definition>
+    clear_policy [-p <vhost>] <name>
+    list_policies [-p <vhost>]
+
+    list_queues [-p <vhost>] [--offline|--online|--local] [<queueinfoitem> ...]
+    list_exchanges [-p <vhost>] [<exchangeinfoitem> ...]
+    list_bindings [-p <vhost>] [<bindinginfoitem> ...]
+    list_connections [<connectioninfoitem> ...]
+    list_channels [<channelinfoitem> ...]
+    list_consumers [-p <vhost>]
+    status
+    node_health_check
+    environment
+    report
+    eval <expr>
+
+    close_connection <connectionpid> <explanation>
+    trace_on [-p <vhost>]
+    trace_off [-p <vhost>]
+    set_vm_memory_high_watermark <fraction>
+    set_vm_memory_high_watermark absolute <memory_limit>
+    set_disk_free_limit <disk_limit>
+    set_disk_free_limit mem_relative <fraction>
+    encode [--decode] [<value>] [<passphrase>] [--list-ciphers] [--list-hashes] 
+[--cipher <cipher>] [--hash <hash>] [--iterations <iterations>]
+    decode [<value>] [<passphrase>][--cipher <cipher>] [--hash <hash>] 
+[--iterations <iterations>]
+    list_hashes
+    list_ciphers
+
+<vhostinfoitem> must be a member of the list [name, tracing].
+
+The list_queues, list_exchanges and list_bindings commands accept an optional 
+virtual host parameter for which to display results. The default value is \"/\".
+
+<queueinfoitem> must be a member of the list [name, durable, auto_delete, 
+arguments, policy, pid, owner_pid, exclusive, exclusive_consumer_pid, 
+exclusive_consumer_tag, messages_ready, messages_unacknowledged, messages, 
+messages_ready_ram, messages_unacknowledged_ram, messages_ram, 
+messages_persistent, message_bytes, message_bytes_ready, 
+message_bytes_unacknowledged, message_bytes_ram, message_bytes_persistent, 
+head_message_timestamp, disk_reads, disk_writes, consumers, 
+consumer_utilisation, memory, slave_pids, synchronised_slave_pids, state].
+
+<exchangeinfoitem> must be a member of the list [name, type, durable, 
+auto_delete, internal, arguments, policy].
+
+<bindinginfoitem> must be a member of the list [source_name, source_kind, 
+destination_name, destination_kind, routing_key, arguments].
+
+<connectioninfoitem> must be a member of the list [pid, name, port, host, 
+peer_port, peer_host, ssl, ssl_protocol, ssl_key_exchange, ssl_cipher, 
+ssl_hash, peer_cert_subject, peer_cert_issuer, peer_cert_validity, state, 
+channels, protocol, auth_mechanism, user, vhost, timeout, frame_max, 
+channel_max, client_properties, recv_oct, recv_cnt, send_oct, send_cnt, 
+send_pend, connected_at].
+
+<channelinfoitem> must be a member of the list [pid, connection, name, number, 
+user, vhost, transactional, confirm, consumer_count, messages_unacknowledged, 
+messages_uncommitted, acks_uncommitted, messages_unconfirmed, prefetch_count, 
+global_prefetch_count].
+
+
+".

src/rabbit_plugins_usage.erl

@@ -0,0 +1,14 @@
+%% Generated, do not edit!
+-module(rabbit_plugins_usage).
+-export([usage/0]).
+usage() -> "Usage:
+rabbitmq-plugins [-n <node>] <command> [<command options>] 
+
+Commands:
+    list [-v] [-m] [-E] [-e] [<pattern>]
+    enable [--offline] [--online] <plugin> ...
+    disable [--offline] [--online] <plugin> ...
+    set [--offline] [--online] <plugin> ...
+
+
+".

src/rabbit_policies.erl

@@ -117,4 +117,3 @@ merge_policy_value(<<"message-ttl">>, Val, OpVal)      -> min(Val, OpVal);
 merge_policy_value(<<"max-length">>, Val, OpVal)       -> min(Val, OpVal);
 merge_policy_value(<<"max-length-bytes">>, Val, OpVal) -> min(Val, OpVal);
 merge_policy_value(<<"expires">>, Val, OpVal)          -> min(Val, OpVal).
-

src/rabbit_priority_queue.erl

@@ -128,11 +128,14 @@ collapse_recovery(QNames, DupNames, Recovery) ->
 priorities(#amqqueue{arguments = Args}) ->
     Ints = [long, short, signedint, byte, unsignedbyte, unsignedshort, unsignedint],
     case rabbit_misc:table_lookup(Args, <<"x-max-priority">>) of
-        {Type, Max} -> case lists:member(Type, Ints) of
-                           false -> none;
-                           true  -> lists:reverse(lists:seq(0, Max))
-                       end;
-        _           -> none
+        {Type, RequestedMax} ->
+            case lists:member(Type, Ints) of
+                false -> none;
+                true  ->
+                    Max = min(RequestedMax, ?MAX_SUPPORTED_PRIORITY),
+                    lists:reverse(lists:seq(0, Max))
+            end;
+        _                    -> none
     end.
 
 %%----------------------------------------------------------------------------

test/priority_queue_SUITE.erl

@@ -17,6 +17,7 @@
 -module(priority_queue_SUITE).
 
 -include_lib("common_test/include/ct.hrl").
+-include_lib("eunit/include/eunit.hrl").
 -include_lib("amqp_client/include/amqp_client.hrl").
 
 -compile(export_all).
@@ -46,7 +47,9 @@ groups() ->
                            simple_order,
                            straight_through,
                            invoke,
-                           gen_server2_stats
+                           gen_server2_stats,
+                           negative_max_priorities,
+                           max_priorities_above_hard_limit
                           ]},
      {cluster_size_3, [], [
                            mirror_queue_auto_ack,
@@ -192,6 +195,28 @@ straight_through(Config) ->
     rabbit_ct_client_helpers:close_connection(Conn),
     passed.
 
+max_priorities_above_hard_limit(Config) ->
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
+    Q = <<"max_priorities_above_hard_limit">>,
+    ?assertExit(
+       {{shutdown, {server_initiated_close, 406, _}}, _},
+       %% Note that lower values (e.g. 300) will cause overflow the byte type here.
+       %% However, values >= 256 would still be rejected when used by
+       %% other clients
+       declare(Ch, Q, 3000)),
+    rabbit_ct_client_helpers:close_connection(Conn),
+    passed.
+
+negative_max_priorities(Config) ->
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
+    Q = <<"negative_max_priorities">>,
+    ?assertExit(
+       {{shutdown, {server_initiated_close, 406, _}}, _},
+       declare(Ch, Q, -10)),
+    rabbit_ct_client_helpers:close_connection(Conn),
+    passed.
+
+
 invoke(Config) ->
     %% Synthetic test to check the invoke callback, as the bug tested here
     %% is only triggered with a race condition.
@@ -669,7 +694,7 @@ get_ok(Ch, Q, Ack, PBin) ->
     {#'basic.get_ok'{delivery_tag = DTag}, #amqp_msg{payload = PBin2}} =
         amqp_channel:call(Ch, #'basic.get'{queue  = Q,
                                            no_ack = Ack =:= no_ack}),
-    PBin = PBin2,
+    ?assertEqual(PBin, PBin2),
     maybe_ack(Ch, Ack, DTag).
 
 get_payload(Ch, Q, Ack, Ps) ->