diff options
| author | Emile Joubert <emile@rabbitmq.com> | 2012-11-22 12:55:30 +0000 |
|---|---|---|
| committer | Emile Joubert <emile@rabbitmq.com> | 2012-11-22 12:55:30 +0000 |
| commit | 2e35c99322e769091fbc582ef0c6354dbd693379 (patch) | |
| tree | f1f1cbaff67173b9fd2f325661aeb498930f4dda | |
| parent | 9ddbcb935c21304fbd40e22714afe62b9c4c4b6b (diff) | |
| parent | 0565c4b1c518d55e6ece9c9bb03662bd54bfdef5 (diff) | |
| download | rabbitmq-server-2e35c99322e769091fbc582ef0c6354dbd693379.tar.gz | |
Merged bug25306
| -rw-r--r-- | src/rabbit_tests.erl | 7 | ||||
| -rw-r--r-- | src/rabbit_variable_queue.erl | 24 |
2 files changed, 20 insertions, 11 deletions
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl index 096f9490..983abf29 100644 --- a/src/rabbit_tests.erl +++ b/src/rabbit_tests.erl @@ -2300,6 +2300,7 @@ test_variable_queue() -> fun test_variable_queue_partial_segments_delta_thing/1, fun test_variable_queue_all_the_bits_not_covered_elsewhere1/1, fun test_variable_queue_all_the_bits_not_covered_elsewhere2/1, + fun test_variable_queue_fold_msg_on_disk/1, fun test_dropwhile/1, fun test_dropwhile_varying_ram_duration/1, fun test_variable_queue_ack_limiting/1, @@ -2515,6 +2516,12 @@ test_variable_queue_all_the_bits_not_covered_elsewhere2(VQ0) -> {empty, VQ8} = rabbit_variable_queue:fetch(false, VQ7), VQ8. +test_variable_queue_fold_msg_on_disk(VQ0) -> + VQ1 = variable_queue_publish(true, 1, VQ0), + {VQ2, AckTags} = variable_queue_fetch(1, true, false, 1, VQ1), + VQ3 = rabbit_variable_queue:fold(fun (_M, _A) -> ok end, VQ2, AckTags), + VQ3. + test_queue_recover() -> Count = 2 * rabbit_queue_index:next_segment_boundary(0), {new, #amqqueue { pid = QPid, name = QName } = Q} = diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl index 8a3fd9d9..6dc65bab 100644 --- a/src/rabbit_variable_queue.erl +++ b/src/rabbit_variable_queue.erl @@ -641,13 +641,12 @@ ack(AckTags, State) -> fold(undefined, State, _AckTags) -> State; fold(MsgFun, State = #vqstate{pending_ack = PA}, AckTags) -> - lists:foldl( - fun(SeqId, State1) -> - {MsgStatus, State2} = - read_msg(gb_trees:get(SeqId, PA), State1), - MsgFun(MsgStatus#msg_status.msg, SeqId), - State2 - end, State, AckTags). + a(lists:foldl(fun(SeqId, State1) -> + {MsgStatus, State2} = + read_msg(gb_trees:get(SeqId, PA), false, State1), + MsgFun(MsgStatus#msg_status.msg, SeqId), + State2 + end, State, AckTags)). requeue(AckTags, #vqstate { delta = Delta, q3 = Q3, @@ -837,6 +836,7 @@ a(State = #vqstate { q1 = Q1, q2 = Q2, delta = Delta, q3 = Q3, q4 = Q4, true = Len >= 0, true = PersistentCount >= 0, true = RamMsgCount >= 0, + true = RamMsgCount =< Len, State. @@ -1062,17 +1062,19 @@ queue_out(State = #vqstate { q4 = Q4 }) -> {{value, MsgStatus}, State #vqstate { q4 = Q4a }} end. +read_msg(MsgStatus, State) -> read_msg(MsgStatus, true, State). + read_msg(MsgStatus = #msg_status { msg = undefined, msg_id = MsgId, is_persistent = IsPersistent }, - State = #vqstate { ram_msg_count = RamMsgCount, - msg_store_clients = MSCState}) -> + CountDiskToRam, State = #vqstate { ram_msg_count = RamMsgCount, + msg_store_clients = MSCState}) -> {{ok, Msg = #basic_message {}}, MSCState1} = msg_store_read(MSCState, IsPersistent, MsgId), {MsgStatus #msg_status { msg = Msg }, - State #vqstate { ram_msg_count = RamMsgCount + 1, + State #vqstate { ram_msg_count = RamMsgCount + one_if(CountDiskToRam), msg_store_clients = MSCState1 }}; -read_msg(MsgStatus, State) -> +read_msg(MsgStatus, _CountDiskToRam, State) -> {MsgStatus, State}. internal_fetch(AckRequired, MsgStatus = #msg_status { |