diff options
| -rw-r--r-- | src/rabbit_basic.erl | 13 | ||||
| -rw-r--r-- | src/rabbit_tests.erl | 129 |
2 files changed, 94 insertions, 48 deletions
diff --git a/src/rabbit_basic.erl b/src/rabbit_basic.erl index f9a8f488af..63d6a4815d 100644 --- a/src/rabbit_basic.erl +++ b/src/rabbit_basic.erl @@ -33,7 +33,7 @@ -include("rabbit.hrl"). -include("rabbit_framing.hrl"). --export([publish/1, message/4, message/5, delivery/4]). +-export([publish/1, message/4, message/5, message/6, delivery/4]). %%---------------------------------------------------------------------------- @@ -44,8 +44,10 @@ -spec(delivery/4 :: (bool(), bool(), maybe(txn()), message()) -> delivery()). -spec(message/4 :: (exchange_name(), routing_key(), binary(), binary()) -> message()). --spec(message/5 :: (exchange_name(), routing_key(), binary(), binary(), guid()) -> - message()). +-spec(message/5 :: (exchange_name(), routing_key(), binary(), binary(), + guid()) -> message()). +-spec(message/6 :: (exchange_name(), routing_key(), binary(), binary(), + guid(), bool()) -> message()). -endif. @@ -69,6 +71,9 @@ message(ExchangeName, RoutingKeyBin, ContentTypeBin, BodyBin) -> message(ExchangeName, RoutingKeyBin, ContentTypeBin, BodyBin, rabbit_guid:guid()). message(ExchangeName, RoutingKeyBin, ContentTypeBin, BodyBin, MsgId) -> + message(ExchangeName, RoutingKeyBin, ContentTypeBin, BodyBin, MsgId, false). + +message(ExchangeName, RoutingKeyBin, ContentTypeBin, BodyBin, MsgId, IsPersistent) -> {ClassId, _MethodId} = rabbit_framing:method_id('basic.publish'), Content = #content{class_id = ClassId, properties = #'P_basic'{content_type = ContentTypeBin}, @@ -78,4 +83,4 @@ message(ExchangeName, RoutingKeyBin, ContentTypeBin, BodyBin, MsgId) -> routing_key = RoutingKeyBin, content = Content, guid = MsgId, - is_persistent = false}. + is_persistent = IsPersistent}. diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl index 37c0121cda..b00331c2d6 100644 --- a/src/rabbit_tests.erl +++ b/src/rabbit_tests.erl @@ -1052,55 +1052,91 @@ rdq_test_mixed_queue_modes() -> passed. rdq_test_mode_conversion_mid_txn() -> + Payload = <<0:(8*256)>>, + MsgIdsA = lists:seq(0,9), + MsgsA = [ rabbit_basic:message(x, <<>>, <<>>, Payload, MsgId, + (0 == MsgId rem 2)) + || MsgId <- MsgIdsA ], + MsgIdsB = lists:seq(10,20), + MsgsB = [ rabbit_basic:message(x, <<>>, <<>>, Payload, MsgId, + (0 == MsgId rem 2)) + || MsgId <- MsgIdsB ], + rdq_virgin(), rdq_start(), - Payload = <<0:(8*256)>>, - {ok, MS} = rabbit_mixed_queue:init(q, true, mixed), - MsgIds = lists:seq(1,10), - {MS2, Msgs} = - lists:foldl( - fun (N, {MS1, Acc}) -> - Msg = rabbit_basic:message(x, <<>>, <<>>, Payload, N), - {ok, MS1a} = rabbit_mixed_queue:tx_publish(Msg, MS1), - {MS1a, [Msg | Acc]} - end, {MS, []}, MsgIds), - MsgsOrdered = lists:reverse(Msgs), - {ok, MS3} = rabbit_mixed_queue:to_disk_only_mode(MsgsOrdered, MS2), - {ok, MS4} = rabbit_mixed_queue:tx_commit(MsgsOrdered, [], MS3), - MS6 = - lists:foldl( - fun (N, MS5) -> - Rem = 10 - N, - {{#basic_message { is_persistent = false }, - false, _AckTag, Rem}, - MS5a} = rabbit_mixed_queue:deliver(MS5), - MS5a - end, MS4, MsgIds), - 0 = rabbit_mixed_queue:length(MS6), - {ok, MS7} = rabbit_mixed_queue:init(q, true, disk), - {MS9, Msgs1} = - lists:foldl( - fun (N, {MS8, Acc}) -> - Msg = rabbit_basic:message(x, <<>>, <<>>, Payload, N), - {ok, MS8a} = rabbit_mixed_queue:tx_publish(Msg, MS8), - {MS8a, [Msg | Acc]} - end, {MS7, []}, MsgIds), - Msgs1Ordered = lists:reverse(Msgs1), - {ok, MS10} = rabbit_mixed_queue:to_mixed_mode(Msgs1Ordered, MS9), - {ok, MS11} = rabbit_mixed_queue:tx_commit(Msgs1Ordered, [], MS10), - MS13 = - lists:foldl( - fun (N, MS12) -> - Rem = 10 - N, - {{#basic_message { is_persistent = false }, - false, _AckTag, Rem}, - MS12a} = rabbit_mixed_queue:deliver(MS12), - MS12a - end, MS11, MsgIds), - 0 = rabbit_mixed_queue:length(MS13), + {ok, MS0} = rabbit_mixed_queue:init(q, true, mixed), + passed = rdq_tx_publish_mixed_alter_commit_get( + MS0, MsgsA, MsgsB, fun rabbit_mixed_queue:to_disk_only_mode/2, commit), + + rdq_stop_virgin_start(), + {ok, MS1} = rabbit_mixed_queue:init(q, true, mixed), + passed = rdq_tx_publish_mixed_alter_commit_get( + MS1, MsgsA, MsgsB, fun rabbit_mixed_queue:to_disk_only_mode/2, cancel), + + + rdq_stop_virgin_start(), + {ok, MS2} = rabbit_mixed_queue:init(q, true, disk), + passed = rdq_tx_publish_mixed_alter_commit_get( + MS2, MsgsA, MsgsB, fun rabbit_mixed_queue:to_mixed_mode/2, commit), + + rdq_stop_virgin_start(), + {ok, MS3} = rabbit_mixed_queue:init(q, true, disk), + passed = rdq_tx_publish_mixed_alter_commit_get( + MS3, MsgsA, MsgsB, fun rabbit_mixed_queue:to_mixed_mode/2, cancel), + rdq_stop(), passed. +rdq_tx_publish_mixed_alter_commit_get(MS0, MsgsA, MsgsB, ChangeFun, CommitOrCancel) -> + 0 = rabbit_mixed_queue:length(MS0), + MS2 = lists:foldl( + fun (Msg, MS1) -> + {ok, MS1a} = rabbit_mixed_queue:publish(Msg, MS1), + MS1a + end, MS0, MsgsA), + Len0 = length(MsgsA), + Len0 = rabbit_mixed_queue:length(MS2), + MS4 = lists:foldl( + fun (Msg, MS3) -> + {ok, MS3a} = rabbit_mixed_queue:tx_publish(Msg, MS3), + MS3a + end, MS2, MsgsB), + Len0 = rabbit_mixed_queue:length(MS4), + {ok, MS5} = ChangeFun(MsgsB, MS4), + Len0 = rabbit_mixed_queue:length(MS5), + {ok, MS9} = + case CommitOrCancel of + commit -> + {ok, MS6} = rabbit_mixed_queue:tx_commit(MsgsB, [], MS5), + Len1 = Len0 + length(MsgsB), + Len1 = rabbit_mixed_queue:length(MS6), + {AckTags, MS8} = + lists:foldl( + fun (Msg, {Acc, MS7}) -> + Rem = Len1 - (Msg #basic_message.guid) - 1, + {{Msg, false, AckTag, Rem}, MS7a} = + rabbit_mixed_queue:deliver(MS7), + {[AckTag | Acc], MS7a} + end, {[], MS6}, MsgsA ++ MsgsB), + 0 = rabbit_mixed_queue:length(MS8), + rabbit_mixed_queue:ack(lists:reverse(AckTags), MS8); + cancel -> + {ok, MS6} = rabbit_mixed_queue:tx_cancel(MsgsB, MS5), + Len0 = rabbit_mixed_queue:length(MS6), + {AckTags, MS8} = + lists:foldl( + fun (Msg, {Acc, MS7}) -> + Rem = Len0 - (Msg #basic_message.guid) - 1, + {{Msg, false, AckTag, Rem}, MS7a} = + rabbit_mixed_queue:deliver(MS7), + {[AckTag | Acc], MS7a} + end, {[], MS6}, MsgsA), + 0 = rabbit_mixed_queue:length(MS8), + rabbit_mixed_queue:ack(lists:reverse(AckTags), MS8) + end, + 0 = rabbit_mixed_queue:length(MS9), + passed. + rdq_time_commands(Funcs) -> lists:foreach(fun (F) -> F() end, Funcs). @@ -1119,3 +1155,8 @@ rdq_start() -> rdq_stop() -> rabbit_disk_queue:stop(), timer:sleep(1000). + +rdq_stop_virgin_start() -> + rdq_stop(), + rdq_virgin(), + rdq_start(). |