diff options
| -rw-r--r-- | src/rabbit_amqqueue.erl | 2 | ||||
| -rw-r--r-- | src/rabbit_amqqueue_process.erl | 16 | ||||
| -rw-r--r-- | src/rabbit_policies.erl | 4 | ||||
| -rw-r--r-- | test/priority_queue_SUITE.erl | 2 | ||||
| -rw-r--r-- | test/simple_ha_SUITE.erl | 2 | ||||
| -rw-r--r-- | test/unit_inbroker_parallel_SUITE.erl | 8 |
6 files changed, 17 insertions, 17 deletions
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl index 592de77c97..ce73460603 100644 --- a/src/rabbit_amqqueue.erl +++ b/src/rabbit_amqqueue.erl @@ -625,7 +625,7 @@ check_dlxrk_arg({Type, _}, _Args) -> {error, {unacceptable_type, Type}}. check_overflow({longstr, Val}, _Args) -> - case lists:member(Val, [<<"drop_head">>, <<"reject_publish">>]) of + case lists:member(Val, [<<"drop-head">>, <<"reject-publish">>]) of true -> ok; false -> {error, invalid_overflow} end; diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index 3cd9f4195d..7721fc70fc 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -80,7 +80,7 @@ %% max length in bytes, if configured max_bytes, %% an action to perform if queue is to be over a limit, - %% can be either drop_head (default) or reject_publish + %% can be either drop-head (default) or reject-publish overflow, %% when policies change, this version helps queue %% determine what previously scheduled/set up state to ignore, @@ -162,7 +162,7 @@ init_state(Q) -> msg_id_to_channel = gb_trees:empty(), status = running, args_policy_version = 0, - overflow = drop_head}, + overflow = 'drop-head'}, rabbit_event:init_stats_timer(State, #q.stats_timer). init_it(Recover, From, State = #q{q = #amqqueue{exclusive_owner = none}}) -> @@ -634,9 +634,9 @@ attempt_delivery(Delivery = #delivery{sender = SenderPid, maybe_deliver_or_enqueue(Delivery, Delivered, State = #q{overflow = Overflow}) -> send_mandatory(Delivery), %% must do this before confirms case {will_overflow(Delivery, State), Overflow} of - {true, reject_publish} -> + {true, 'reject-publish'} -> %% Drop publish and nack to publisher - nack_publish_no_space(Delivery, Delivered, State); + send_reject_publish(Delivery, Delivered, State); _ -> %% Enqueue and maybe drop head later deliver_or_enqueue(Delivery, Delivered, State) @@ -686,9 +686,9 @@ deliver_or_enqueue(Delivery = #delivery{message = Message, maybe_drop_head(State = #q{max_length = undefined, max_bytes = undefined}) -> {false, State}; -maybe_drop_head(State = #q{overflow = reject_publish}) -> +maybe_drop_head(State = #q{overflow = 'reject-publish'}) -> {false, State}; -maybe_drop_head(State = #q{overflow = drop_head}) -> +maybe_drop_head(State = #q{overflow = 'drop-head'}) -> maybe_drop_head(false, State). maybe_drop_head(AlreadyDropped, State = #q{backing_queue = BQ, @@ -707,7 +707,7 @@ maybe_drop_head(AlreadyDropped, State = #q{backing_queue = BQ, {AlreadyDropped, State} end. -nack_publish_no_space(#delivery{confirm = true, +send_reject_publish(#delivery{confirm = true, sender = SenderPid, msg_seq_no = MsgSeqNo} = Delivery, _Delivered, @@ -717,7 +717,7 @@ nack_publish_no_space(#delivery{confirm = true, {BQS1, MTC1} = discard(Delivery, BQ, BQS, MTC), gen_server2:cast(SenderPid, {reject_publish, MsgSeqNo, self()}), State#q{ backing_queue_state = BQS1, msg_id_to_channel = MTC1 }; -nack_publish_no_space(#delivery{confirm = false}, +send_reject_publish(#delivery{confirm = false}, _Delivered, State) -> State. diff --git a/src/rabbit_policies.erl b/src/rabbit_policies.erl index 3fadfc82e1..f48189b210 100644 --- a/src/rabbit_policies.erl +++ b/src/rabbit_policies.erl @@ -106,9 +106,9 @@ validate_policy0(<<"queue-mode">>, <<"lazy">>) -> ok; validate_policy0(<<"queue-mode">>, Value) -> {error, "~p is not a valid queue-mode value", [Value]}; -validate_policy0(<<"overflow">>, <<"drop_head">>) -> +validate_policy0(<<"overflow">>, <<"drop-head">>) -> ok; -validate_policy0(<<"overflow">>, <<"reject_publish">>) -> +validate_policy0(<<"overflow">>, <<"reject-publish">>) -> ok; validate_policy0(<<"overflow">>, Value) -> {error, "~p is not a valid overflow value", [Value]}. diff --git a/test/priority_queue_SUITE.erl b/test/priority_queue_SUITE.erl index eb781ffedf..a1ae66dbbb 100644 --- a/test/priority_queue_SUITE.erl +++ b/test/priority_queue_SUITE.erl @@ -311,7 +311,7 @@ reject(Config) -> {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0), Q = <<"reject-queue">>, declare(Ch, Q, [{<<"x-max-length">>, long, 4}, - {<<"x-overflow">>, longstr, <<"reject_publish">>} + {<<"x-overflow">>, longstr, <<"reject-publish">>} | arguments(3)]), publish(Ch, Q, [1, 2, 3, 1, 2, 3, 1, 2, 3]), %% First 4 messages are published, all others are discarded. diff --git a/test/simple_ha_SUITE.erl b/test/simple_ha_SUITE.erl index 297b53b713..f3af62d43a 100644 --- a/test/simple_ha_SUITE.erl +++ b/test/simple_ha_SUITE.erl @@ -230,7 +230,7 @@ rejecets_survive(Config, DeathFun) -> auto_delete = false, durable = true, arguments = [{<<"x-max-length">>, long, 1}, - {<<"x-overflow">>, longstr, <<"reject_publish">>}]}), + {<<"x-overflow">>, longstr, <<"reject-publish">>}]}), Payload = <<"there can be only one">>, amqp_channel:call(Node1Channel, #'basic.publish'{routing_key = Queue}, diff --git a/test/unit_inbroker_parallel_SUITE.erl b/test/unit_inbroker_parallel_SUITE.erl index 951cf6f213..d9ea349f1b 100644 --- a/test/unit_inbroker_parallel_SUITE.erl +++ b/test/unit_inbroker_parallel_SUITE.erl @@ -1050,7 +1050,7 @@ max_length_drop_head(Config) -> MaxLengthArgs = [{<<"x-max-length">>, long, 1}], MaxLengthBytesArgs = [{<<"x-max-length-bytes">>, long, 100}], - OverflowArgs = [{<<"x-overflow">>, longstr, <<"drop_head">>}], + OverflowArgs = [{<<"x-overflow">>, longstr, <<"drop-head">>}], amqp_channel:call(Ch, #'queue.delete'{queue = QName}), amqp_channel:call(Ch, #'queue.delete'{queue = QNameDefault}), amqp_channel:call(Ch, #'queue.delete'{queue = QNameBytes}), @@ -1076,7 +1076,7 @@ max_length_reject_confirm(Config) -> QNameBytes = <<"max_length_bytes_reject_queue">>, MaxLengthArgs = [{<<"x-max-length">>, long, 1}], MaxLengthBytesArgs = [{<<"x-max-length-bytes">>, long, 100}], - OverflowArgs = [{<<"x-overflow">>, longstr, <<"reject_publish">>}], + OverflowArgs = [{<<"x-overflow">>, longstr, <<"reject-publish">>}], amqp_channel:call(Ch, #'queue.delete'{queue = QName}), amqp_channel:call(Ch, #'queue.delete'{queue = QNameBytes}), #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = QName, arguments = MaxLengthArgs ++ OverflowArgs}), @@ -1099,12 +1099,12 @@ max_length_drop_publish(Config) -> QNameBytes = <<"max_length_bytes_drop_publish_queue">>, MaxLengthArgs = [{<<"x-max-length">>, long, 1}], MaxLengthBytesArgs = [{<<"x-max-length-bytes">>, long, 100}], - OverflowArgs = [{<<"x-overflow">>, longstr, <<"reject_publish">>}], + OverflowArgs = [{<<"x-overflow">>, longstr, <<"reject-publish">>}], amqp_channel:call(Ch, #'queue.delete'{queue = QName}), amqp_channel:call(Ch, #'queue.delete'{queue = QNameBytes}), #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = QName, arguments = MaxLengthArgs ++ OverflowArgs}), #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = QNameBytes, arguments = MaxLengthBytesArgs ++ OverflowArgs}), - %% If confirms are not enable, publishes will still be dropped in reject_publish mode. + %% If confirms are not enable, publishes will still be dropped in reject-publish mode. check_max_length_drops_publish(Config, QName, Ch, <<"1">>, <<"2">>, <<"3">>), %% 80 bytes payload |