diff options
author | Tim Watson <tim@rabbitmq.com> | 2012-11-29 13:25:49 +0000 |
---|---|---|
committer | Tim Watson <tim@rabbitmq.com> | 2012-11-29 13:25:49 +0000 |
commit | 5f0ddb28ffe3a504987461df2d9f2bdc4aad2290 (patch) | |
tree | 39d149857f48a068e4539ec4f548e728a7054990 | |
parent | 76e832a6c6e451d049a1c5439251179444bbc966 (diff) | |
parent | ac7c973b46b9b1deffdfa7acde907e5e4c48cbab (diff) | |
download | rabbitmq-server-5f0ddb28ffe3a504987461df2d9f2bdc4aad2290.tar.gz |
merge heads
-rw-r--r-- | README | 2 | ||||
-rw-r--r-- | include/rabbit.hrl | 3 | ||||
-rw-r--r-- | src/rabbit.erl | 8 | ||||
-rw-r--r-- | src/rabbit_amqqueue.erl | 4 | ||||
-rw-r--r-- | src/rabbit_amqqueue_process.erl | 58 | ||||
-rw-r--r-- | src/rabbit_backing_queue.erl | 11 | ||||
-rw-r--r-- | src/rabbit_backing_queue_qc.erl | 4 | ||||
-rw-r--r-- | src/rabbit_mirror_queue_master.erl | 39 | ||||
-rw-r--r-- | src/rabbit_mirror_queue_misc.erl | 16 | ||||
-rw-r--r-- | src/rabbit_mirror_queue_slave.erl | 3 | ||||
-rw-r--r-- | src/rabbit_policy.erl | 4 | ||||
-rw-r--r-- | src/rabbit_tests.erl | 5 | ||||
-rw-r--r-- | src/rabbit_variable_queue.erl | 55 |
13 files changed, 92 insertions, 120 deletions
@@ -1 +1 @@ -Please see http://www.rabbitmq.com/build-server.html for build instructions.
\ No newline at end of file +Please see http://www.rabbitmq.com/build-server.html for build instructions. diff --git a/include/rabbit.hrl b/include/rabbit.hrl index b2832b45..0ccb80bf 100644 --- a/include/rabbit.hrl +++ b/include/rabbit.hrl @@ -78,8 +78,7 @@ -record(event, {type, props, timestamp}). --record(message_properties, {expiry, needs_confirming = false, - delivered = false}). +-record(message_properties, {expiry, needs_confirming = false}). -record(plugin, {name, %% atom() version, %% string() diff --git a/src/rabbit.erl b/src/rabbit.erl index c3a6d283..7b8348fc 100644 --- a/src/rabbit.erl +++ b/src/rabbit.erl @@ -400,13 +400,11 @@ status() -> is_running() -> is_running(node()). -is_running(Node) -> - rabbit_nodes:is_running(Node, rabbit). +is_running(Node) -> rabbit_nodes:is_running(Node, rabbit). environment() -> - lists:keysort( - 1, [P || P = {K, _} <- application:get_all_env(rabbit), - K =/= default_pass]). + lists:keysort(1, [P || P = {K, _} <- application:get_all_env(rabbit), + K =/= default_pass]). rotate_logs(BinarySuffix) -> Suffix = binary_to_list(BinarySuffix), diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl index be7c7867..7827b839 100644 --- a/src/rabbit_amqqueue.erl +++ b/src/rabbit_amqqueue.erl @@ -587,8 +587,8 @@ set_ram_duration_target(QPid, Duration) -> set_maximum_since_use(QPid, Age) -> gen_server2:cast(QPid, {set_maximum_since_use, Age}). -start_mirroring(QPid) -> ok = delegate_call(QPid, start_mirroring). -stop_mirroring(QPid) -> ok = delegate_call(QPid, stop_mirroring). +start_mirroring(QPid) -> ok = delegate_cast(QPid, start_mirroring). +stop_mirroring(QPid) -> ok = delegate_cast(QPid, stop_mirroring). on_node_down(Node) -> rabbit_misc:execute_mnesia_tx_with_tail( diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index 5ddafba8..74717ace 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -120,11 +120,10 @@ info_keys() -> ?INFO_KEYS. init(Q) -> process_flag(trap_exit, true), - State = #q{q = Q#amqqueue{pid = self()}, exclusive_consumer = none, has_had_consumers = false, - backing_queue = backing_queue_module(Q), + backing_queue = undefined, backing_queue_state = undefined, active_consumers = queue:new(), expires = undefined, @@ -191,7 +190,7 @@ code_change(_OldVsn, State, _Extra) -> %%---------------------------------------------------------------------------- declare(Recover, From, State = #q{q = Q, - backing_queue = BQ, + backing_queue = undefined, backing_queue_state = undefined}) -> case rabbit_amqqueue:internal_declare(Q, Recover =/= new) of #amqqueue{} = Q1 -> @@ -203,9 +202,11 @@ declare(Recover, From, State = #q{q = Q, ok = rabbit_memory_monitor:register( self(), {rabbit_amqqueue, set_ram_duration_target, [self()]}), + BQ = backing_queue_module(Q1), BQS = bq_init(BQ, Q, Recover), recovery_barrier(Recover), - State1 = process_args(State#q{backing_queue_state = BQS}), + State1 = process_args(State#q{backing_queue = BQ, + backing_queue_state = BQS}), rabbit_event:notify(queue_created, infos(?CREATION_EVENT_KEYS, State1)), rabbit_event:if_enabled(State1, #q.stats_timer, @@ -540,8 +541,8 @@ run_message_queue(State) -> State2. attempt_delivery(Delivery = #delivery{sender = SenderPid, message = Message}, - Props = #message_properties{delivered = Delivered}, - State = #q{backing_queue = BQ, backing_queue_state = BQS}) -> + Props, Delivered, State = #q{backing_queue = BQ, + backing_queue_state = BQS}) -> case BQ:is_duplicate(Message, BQS) of {false, BQS1} -> deliver_msgs_to_consumers( @@ -563,15 +564,15 @@ attempt_delivery(Delivery = #delivery{sender = SenderPid, message = Message}, deliver_or_enqueue(Delivery = #delivery{message = Message, sender = SenderPid}, Delivered, State) -> {Confirm, State1} = send_or_record_confirm(Delivery, State), - Props = message_properties(Message, Confirm, Delivered, State), - case attempt_delivery(Delivery, Props, State1) of + Props = message_properties(Message, Confirm, State), + case attempt_delivery(Delivery, Props, Delivered, State1) of {true, State2} -> State2; %% The next one is an optimisation {false, State2 = #q{ttl = 0, dlx = undefined}} -> discard(Delivery, State2); {false, State2 = #q{backing_queue = BQ, backing_queue_state = BQS}} -> - BQS1 = BQ:publish(Message, Props, SenderPid, BQS), + BQS1 = BQ:publish(Message, Props, Delivered, SenderPid, BQS), ensure_ttl_timer(Props#message_properties.expiry, State2#q{backing_queue_state = BQS1}) end. @@ -704,10 +705,9 @@ subtract_acks(ChPid, AckTags, State, Fun) -> Fun(State) end. -message_properties(Message, Confirm, Delivered, #q{ttl = TTL}) -> +message_properties(Message, Confirm, #q{ttl = TTL}) -> #message_properties{expiry = calculate_msg_expiry(Message, TTL), - needs_confirming = Confirm == eventually, - delivered = Delivered}. + needs_confirming = Confirm == eventually}. calculate_msg_expiry(#basic_message{content = Content}, TTL) -> #content{properties = Props} = @@ -1153,23 +1153,6 @@ handle_call({requeue, AckTags, ChPid}, From, State) -> gen_server2:reply(From, ok), noreply(requeue(AckTags, ChPid, State)); -handle_call(start_mirroring, _From, State = #q{backing_queue = BQ, - backing_queue_state = BQS}) -> - %% lookup again to get policy for init_with_existing_bq - {ok, Q} = rabbit_amqqueue:lookup(qname(State)), - true = BQ =/= rabbit_mirror_queue_master, %% assertion - BQ1 = rabbit_mirror_queue_master, - BQS1 = BQ1:init_with_existing_bq(Q, BQ, BQS), - reply(ok, State#q{backing_queue = BQ1, - backing_queue_state = BQS1}); - -handle_call(stop_mirroring, _From, State = #q{backing_queue = BQ, - backing_queue_state = BQS}) -> - BQ = rabbit_mirror_queue_master, %% assertion - {BQ1, BQS1} = BQ:stop_mirroring(BQS), - reply(ok, State#q{backing_queue = BQ1, - backing_queue_state = BQS1}); - handle_call(force_event_refresh, _From, State = #q{exclusive_consumer = Exclusive}) -> rabbit_event:notify(queue_created, infos(?CREATION_EVENT_KEYS, State)), @@ -1301,6 +1284,23 @@ handle_cast({dead_letter, Msgs, Reason}, State = #q{dlx = XName}) -> cleanup_after_confirm([AckTag || {_, AckTag} <- Msgs], State) end; +handle_cast(start_mirroring, State = #q{backing_queue = BQ, + backing_queue_state = BQS}) -> + %% lookup again to get policy for init_with_existing_bq + {ok, Q} = rabbit_amqqueue:lookup(qname(State)), + true = BQ =/= rabbit_mirror_queue_master, %% assertion + BQ1 = rabbit_mirror_queue_master, + BQS1 = BQ1:init_with_existing_bq(Q, BQ, BQS), + noreply(State#q{backing_queue = BQ1, + backing_queue_state = BQS1}); + +handle_cast(stop_mirroring, State = #q{backing_queue = BQ, + backing_queue_state = BQS}) -> + BQ = rabbit_mirror_queue_master, %% assertion + {BQ1, BQS1} = BQ:stop_mirroring(BQS), + noreply(State#q{backing_queue = BQ1, + backing_queue_state = BQS1}); + handle_cast(wake_up, State) -> noreply(State). diff --git a/src/rabbit_backing_queue.erl b/src/rabbit_backing_queue.erl index 9e99ca5e..e2945e1d 100644 --- a/src/rabbit_backing_queue.erl +++ b/src/rabbit_backing_queue.erl @@ -78,8 +78,8 @@ %% Publish a message. -callback publish(rabbit_types:basic_message(), - rabbit_types:message_properties(), pid(), state()) -> - state(). + rabbit_types:message_properties(), boolean(), pid(), + state()) -> state(). %% Called for messages which have already been passed straight %% out to a client. The queue will be empty for these calls @@ -157,8 +157,9 @@ %% Fold over all the messages in a queue and return the accumulated %% results, leaving the queue undisturbed. --callback fold(fun((rabbit_types:basic_message(), A) -> A), A, state()) - -> {A, state()}. +-callback fold(fun((rabbit_types:basic_message(), + rabbit_types:message_properties(), A) -> A), + A, state()) -> {A, state()}. %% How long is my queue? -callback len(state()) -> non_neg_integer(). @@ -219,7 +220,7 @@ behaviour_info(callbacks) -> [{start, 1}, {stop, 0}, {init, 3}, {terminate, 2}, - {delete_and_terminate, 2}, {purge, 1}, {publish, 4}, + {delete_and_terminate, 2}, {purge, 1}, {publish, 5}, {publish_delivered, 4}, {discard, 3}, {drain_confirmed, 1}, {dropwhile, 3}, {fetch, 2}, {ack, 2}, {foreach_ack, 3}, {requeue, 2}, {fold, 3}, {len, 1}, {is_empty, 1}, {depth, 1}, {set_ram_duration_target, 2}, diff --git a/src/rabbit_backing_queue_qc.erl b/src/rabbit_backing_queue_qc.erl index 03808859..f258e15e 100644 --- a/src/rabbit_backing_queue_qc.erl +++ b/src/rabbit_backing_queue_qc.erl @@ -331,8 +331,8 @@ postcondition(S, {call, ?BQMOD, drain_confirmed, _Args}, Res) -> postcondition(S, {call, ?BQMOD, fold, _Args}, {Res, _BQ}) -> #state{messages = Messages} = S, - lists:foldl(fun ({_SeqId, {_MsgProps, Msg}}, Acc) -> - foldfun(Msg, Acc) + lists:foldl(fun ({_SeqId, {MsgProps, Msg}}, Acc) -> + foldfun(Msg, MsgProps, Acc) end, foldacc(), gb_trees:to_list(Messages)) =:= Res; postcondition(#state{bqstate = BQ, len = Len}, {call, _M, _F, _A}, _Res) -> diff --git a/src/rabbit_mirror_queue_master.erl b/src/rabbit_mirror_queue_master.erl index 8fcd1893..c8a361b1 100644 --- a/src/rabbit_mirror_queue_master.erl +++ b/src/rabbit_mirror_queue_master.erl @@ -17,7 +17,7 @@ -module(rabbit_mirror_queue_master). -export([init/3, terminate/2, delete_and_terminate/2, - purge/1, publish/4, publish_delivered/4, + purge/1, publish/5, publish_delivered/4, discard/3, fetch/2, drop/2, ack/2, requeue/2, fold/3, len/1, is_empty/1, depth/1, drain_confirmed/1, dropwhile/3, set_ram_duration_target/2, ram_duration/1, @@ -38,7 +38,6 @@ coordinator, backing_queue, backing_queue_state, - set_delivered, seen_status, confirmed, ack_msg_id, @@ -55,7 +54,6 @@ coordinator :: pid(), backing_queue :: atom(), backing_queue_state :: any(), - set_delivered :: non_neg_integer(), seen_status :: dict(), confirmed :: [rabbit_guid:guid()], ack_msg_id :: dict(), @@ -114,7 +112,6 @@ init_with_existing_bq(Q = #amqqueue{name = QName}, BQ, BQS) -> coordinator = CPid, backing_queue = BQ, backing_queue_state = BQS, - set_delivered = 0, seen_status = dict:new(), confirmed = [], ack_msg_id = dict:new(), @@ -136,8 +133,8 @@ terminate({shutdown, dropped} = Reason, %% in without this node being restarted. Thus we must do the full %% blown delete_and_terminate now, but only locally: we do not %% broadcast delete_and_terminate. - State #state { backing_queue_state = BQ:delete_and_terminate(Reason, BQS), - set_delivered = 0 }; + State#state{backing_queue_state = BQ:delete_and_terminate(Reason, BQS)}; + terminate(Reason, State = #state { backing_queue = BQ, backing_queue_state = BQS }) -> %% Backing queue termination. The queue is going down but @@ -148,8 +145,7 @@ terminate(Reason, delete_and_terminate(Reason, State = #state { backing_queue = BQ, backing_queue_state = BQS }) -> stop_all_slaves(Reason, State), - State #state { backing_queue_state = BQ:delete_and_terminate(Reason, BQS), - set_delivered = 0 }. + State#state{backing_queue_state = BQ:delete_and_terminate(Reason, BQS)}. stop_all_slaves(Reason, #state{gm = GM}) -> Info = gm:info(GM), @@ -175,17 +171,16 @@ purge(State = #state { gm = GM, backing_queue_state = BQS }) -> ok = gm:broadcast(GM, {drop, 0, BQ:len(BQS), false}), {Count, BQS1} = BQ:purge(BQS), - {Count, State #state { backing_queue_state = BQS1, - set_delivered = 0 }}. + {Count, State #state { backing_queue_state = BQS1 }}. -publish(Msg = #basic_message { id = MsgId }, MsgProps, ChPid, +publish(Msg = #basic_message { id = MsgId }, MsgProps, IsDelivered, ChPid, State = #state { gm = GM, seen_status = SS, backing_queue = BQ, backing_queue_state = BQS }) -> false = dict:is_key(MsgId, SS), %% ASSERTION ok = gm:broadcast(GM, {publish, ChPid, MsgProps, Msg}), - BQS1 = BQ:publish(Msg, MsgProps, ChPid, BQS), + BQS1 = BQ:publish(Msg, MsgProps, IsDelivered, ChPid, BQS), ensure_monitoring(ChPid, State #state { backing_queue_state = BQS1 }). publish_delivered(Msg = #basic_message { id = MsgId }, MsgProps, @@ -224,7 +219,6 @@ discard(MsgId, ChPid, State = #state { gm = GM, dropwhile(Pred, AckRequired, State = #state{gm = GM, backing_queue = BQ, - set_delivered = SetDelivered, backing_queue_state = BQS }) -> Len = BQ:len(BQS), {Next, Msgs, BQS1} = BQ:dropwhile(Pred, AckRequired, BQS), @@ -234,9 +228,7 @@ dropwhile(Pred, AckRequired, 0 -> ok; _ -> ok = gm:broadcast(GM, {drop, Len1, Dropped, AckRequired}) end, - SetDelivered1 = lists:max([0, SetDelivered - Dropped]), - {Next, Msgs, State #state { backing_queue_state = BQS1, - set_delivered = SetDelivered1 } }. + {Next, Msgs, State #state { backing_queue_state = BQS1 } }. drain_confirmed(State = #state { backing_queue = BQ, backing_queue_state = BQS, @@ -269,16 +261,14 @@ drain_confirmed(State = #state { backing_queue = BQ, confirmed = [] }}. fetch(AckRequired, State = #state { backing_queue = BQ, - backing_queue_state = BQS, - set_delivered = SetDelivered }) -> + backing_queue_state = BQS }) -> {Result, BQS1} = BQ:fetch(AckRequired, BQS), State1 = State #state { backing_queue_state = BQS1 }, case Result of empty -> {Result, State1}; - {Message, IsDelivered, AckTag} -> - {{Message, IsDelivered orelse SetDelivered > 0, AckTag}, - drop(Message#basic_message.id, AckTag, State1)} + {#basic_message{id = MsgId}, _IsDelivered, AckTag} -> + {Result, drop(MsgId, AckTag, State1)} end. drop(AckRequired, State = #state { backing_queue = BQ, @@ -416,7 +406,6 @@ promote_backing_queue_state(CPid, BQ, BQS, GM, AckTags, SeenStatus, KS) -> coordinator = CPid, backing_queue = BQ, backing_queue_state = BQS1, - set_delivered = Len, seen_status = SeenStatus, confirmed = [], ack_msg_id = dict:new(), @@ -451,14 +440,12 @@ depth_fun() -> %% Helpers %% --------------------------------------------------------------------------- -drop(MsgId, AckTag, State = #state { set_delivered = SetDelivered, - ack_msg_id = AM, +drop(MsgId, AckTag, State = #state { ack_msg_id = AM, gm = GM, backing_queue = BQ, backing_queue_state = BQS }) -> ok = gm:broadcast(GM, {drop, BQ:len(BQS), 1, AckTag =/= undefined}), - State #state { set_delivered = lists:max([0, SetDelivered - 1]), - ack_msg_id = maybe_store_acktag(AckTag, MsgId, AM) }. + State #state { ack_msg_id = maybe_store_acktag(AckTag, MsgId, AM) }. maybe_store_acktag(undefined, _MsgId, AM) -> AM; maybe_store_acktag(AckTag, MsgId, AM) -> dict:store(AckTag, MsgId, AM). diff --git a/src/rabbit_mirror_queue_misc.erl b/src/rabbit_mirror_queue_misc.erl index 2b3bd027..58f20476 100644 --- a/src/rabbit_mirror_queue_misc.erl +++ b/src/rabbit_mirror_queue_misc.erl @@ -133,11 +133,11 @@ on_node_up() -> end end, [], rabbit_queue) end), - [{ok, _} = add_mirror(QName, node()) || QName <- QNames], + [add_mirror(QName, node()) || QName <- QNames], ok. drop_mirrors(QName, Nodes) -> - [{ok, _} = drop_mirror(QName, Node) || Node <- Nodes], + [drop_mirror(QName, Node) || Node <- Nodes], ok. drop_mirror(QName, MirrorNode) -> @@ -159,7 +159,7 @@ drop_mirror(QName, MirrorNode) -> end). add_mirrors(QName, Nodes) -> - [{ok, _} = add_mirror(QName, Node) || Node <- Nodes], + [add_mirror(QName, Node) || Node <- Nodes], ok. add_mirror(QName, MirrorNode) -> @@ -183,15 +183,7 @@ start_child(Name, MirrorNode, Q) -> fun () -> rabbit_mirror_queue_slave_sup:start_child(MirrorNode, [Q]) end) of - {ok, undefined} -> - %% this means the mirror process was - %% already running on the given node. - {ok, already_mirrored}; - {ok, down} -> - %% Node went down between us deciding to start a mirror - %% and actually starting it. Which is fine. - {ok, node_down}; - {ok, SPid} -> + {ok, SPid} when is_pid(SPid) -> rabbit_log:info("Adding mirror of ~s on node ~p: ~p~n", [rabbit_misc:rs(Name), MirrorNode, SPid]), {ok, started}; diff --git a/src/rabbit_mirror_queue_slave.erl b/src/rabbit_mirror_queue_slave.erl index cb7a2135..9354f485 100644 --- a/src/rabbit_mirror_queue_slave.erl +++ b/src/rabbit_mirror_queue_slave.erl @@ -318,7 +318,6 @@ prioritise_cast(Msg, _State) -> {set_maximum_since_use, _Age} -> 8; {run_backing_queue, _Mod, _Fun} -> 6; {gm, _Msg} -> 5; - {post_commit, _Txn, _AckTags} -> 4; _ -> 0 end. @@ -703,7 +702,7 @@ process_instruction({publish, ChPid, MsgProps, Msg = #basic_message { id = MsgId }}, State) -> State1 = #state { backing_queue = BQ, backing_queue_state = BQS } = publish_or_discard(published, ChPid, MsgId, State), - BQS1 = BQ:publish(Msg, MsgProps, ChPid, BQS), + BQS1 = BQ:publish(Msg, MsgProps, true, ChPid, BQS), {ok, State1 #state { backing_queue_state = BQS1 }}; process_instruction({publish_delivered, ChPid, MsgProps, Msg = #basic_message { id = MsgId }}, State) -> diff --git a/src/rabbit_policy.erl b/src/rabbit_policy.erl index 2717cc92..2c997f16 100644 --- a/src/rabbit_policy.erl +++ b/src/rabbit_policy.erl @@ -166,8 +166,8 @@ update_policies(VHost) -> [update_queue(Q, Policies) || Q <- rabbit_amqqueue:list(VHost)]} end), - [notify(X) || X <- Xs], - [notify(Q) || Q <- Qs], + [catch notify(X) || X <- Xs], + [catch notify(Q) || Q <- Qs], ok. update_exchange(X = #exchange{name = XName, policy = OldPolicy}, Policies) -> diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl index 81180ebe..bb4ddceb 100644 --- a/src/rabbit_tests.erl +++ b/src/rabbit_tests.erl @@ -2230,7 +2230,7 @@ variable_queue_publish(IsPersistent, Count, PropFun, PayloadFun, VQ) -> false -> 1 end}, PayloadFun(N)), - PropFun(N, #message_properties{}), self(), VQN) + PropFun(N, #message_properties{}), false, self(), VQN) end, VQ, lists:seq(1, Count)). variable_queue_fetch(Count, IsPersistent, IsDelivered, Len, VQ) -> @@ -2319,7 +2319,8 @@ test_variable_queue_fold(VQ0) -> VQ1 = rabbit_variable_queue:set_ram_duration_target(0, VQ0), VQ2 = variable_queue_publish( true, Count, fun (_, P) -> P end, fun erlang:term_to_binary/1, VQ1), - {Acc, VQ3} = rabbit_variable_queue:fold(fun (M, A) -> [M | A] end, [], VQ2), + {Acc, VQ3} = rabbit_variable_queue:fold( + fun (M, _, A) -> [M | A] end, [], VQ2), true = [term_to_binary(N) || N <- lists:seq(Count, 1, -1)] == [list_to_binary(lists:reverse(P)) || #basic_message{ content = #content{ payload_fragments_rev = P}} <- diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl index e2566e10..7fbac782 100644 --- a/src/rabbit_variable_queue.erl +++ b/src/rabbit_variable_queue.erl @@ -17,7 +17,7 @@ -module(rabbit_variable_queue). -export([init/3, terminate/2, delete_and_terminate/2, purge/1, - publish/4, publish_delivered/4, discard/3, drain_confirmed/1, + publish/5, publish_delivered/4, discard/3, drain_confirmed/1, dropwhile/3, fetch/2, drop/2, ack/2, requeue/2, fold/3, len/1, is_empty/1, depth/1, set_ram_duration_target/2, ram_duration/1, needs_timeout/1, timeout/1, handle_pre_hibernate/1, status/1, invoke/3, @@ -520,16 +520,16 @@ purge(State = #vqstate { q4 = Q4, publish(Msg = #basic_message { is_persistent = IsPersistent, id = MsgId }, MsgProps = #message_properties { needs_confirming = NeedsConfirming }, - _ChPid, State = #vqstate { q1 = Q1, q3 = Q3, q4 = Q4, - next_seq_id = SeqId, - len = Len, - in_counter = InCount, - persistent_count = PCount, - durable = IsDurable, - ram_msg_count = RamMsgCount, - unconfirmed = UC }) -> + IsDelivered, _ChPid, State = #vqstate { q1 = Q1, q3 = Q3, q4 = Q4, + next_seq_id = SeqId, + len = Len, + in_counter = InCount, + persistent_count = PCount, + durable = IsDurable, + ram_msg_count = RamMsgCount, + unconfirmed = UC }) -> IsPersistent1 = IsDurable andalso IsPersistent, - MsgStatus = msg_status(IsPersistent1, SeqId, Msg, MsgProps), + MsgStatus = msg_status(IsPersistent1, IsDelivered, SeqId, Msg, MsgProps), {MsgStatus1, State1} = maybe_write_to_disk(false, false, MsgStatus, State), State2 = case ?QUEUE:is_empty(Q3) of false -> State1 #vqstate { q1 = ?QUEUE:in(m(MsgStatus1), Q1) }; @@ -556,8 +556,7 @@ publish_delivered(Msg = #basic_message { is_persistent = IsPersistent, durable = IsDurable, unconfirmed = UC }) -> IsPersistent1 = IsDurable andalso IsPersistent, - MsgStatus = (msg_status(IsPersistent1, SeqId, Msg, MsgProps)) - #msg_status { is_delivered = true }, + MsgStatus = msg_status(IsPersistent1, true, SeqId, Msg, MsgProps), {MsgStatus1, State1} = maybe_write_to_disk(false, false, MsgStatus, State), State2 = record_pending_ack(m(MsgStatus1), State1), PCount1 = PCount + one_if(IsPersistent1), @@ -685,9 +684,9 @@ fold(Fun, Acc, #vqstate { q1 = Q1, q3 = Q3, q4 = Q4 } = State) -> QFun = fun(MsgStatus, {Acc0, State0}) -> - {#msg_status { msg = Msg }, State1 } = + {#msg_status { msg = Msg, msg_props = MsgProps }, State1 } = read_msg(MsgStatus, false, State0), - {Fun(Msg, Acc0), State1} + {Fun(Msg, MsgProps, Acc0), State1} end, {Acc1, State1} = ?QUEUE:foldl(QFun, {Acc, State}, Q4), {Acc2, State2} = ?QUEUE:foldl(QFun, {Acc1, State1}, Q3), @@ -891,11 +890,10 @@ gb_sets_maybe_insert(false, _Val, Set) -> Set; %% when requeueing, we re-add a msg_id to the unconfirmed set gb_sets_maybe_insert(true, Val, Set) -> gb_sets:add(Val, Set). -msg_status(IsPersistent, SeqId, Msg = #basic_message { id = MsgId }, - MsgProps = #message_properties { delivered = Delivered }) -> - %% TODO would it make sense to remove #msg_status.is_delivered? +msg_status(IsPersistent, IsDelivered, SeqId, + Msg = #basic_message { id = MsgId }, MsgProps) -> #msg_status { seq_id = SeqId, msg_id = MsgId, msg = Msg, - is_persistent = IsPersistent, is_delivered = Delivered, + is_persistent = IsPersistent, is_delivered = IsDelivered, msg_on_disk = false, index_on_disk = false, msg_props = MsgProps }. @@ -1249,17 +1247,15 @@ maybe_write_to_disk(ForceMsg, ForceIndex, MsgStatus, %% Internal gubbins for acks %%---------------------------------------------------------------------------- -record_pending_ack(#msg_status { seq_id = SeqId, - msg_on_disk = MsgOnDisk } = MsgStatus, +record_pending_ack(#msg_status { seq_id = SeqId, msg = Msg } = MsgStatus, State = #vqstate { pending_ack = PA, ram_ack_index = RAI, ack_in_counter = AckInCount}) -> - {AckEntry, RAI1} = - case MsgOnDisk of - true -> {m(trim_msg_status(MsgStatus)), RAI}; - false -> {MsgStatus, gb_sets:insert(SeqId, RAI)} - end, - State #vqstate { pending_ack = gb_trees:insert(SeqId, AckEntry, PA), + RAI1 = case Msg of + undefined -> RAI; + _ -> gb_sets:insert(SeqId, RAI) + end, + State #vqstate { pending_ack = gb_trees:insert(SeqId, MsgStatus, PA), ram_ack_index = RAI1, ack_in_counter = AckInCount + 1}. @@ -1451,11 +1447,11 @@ delta_fold(Fun, Acc, DeltaSeqId, DeltaSeqIdEnd, {List, IndexState1} = rabbit_queue_index:read(DeltaSeqId, DeltaSeqId1, IndexState), {Acc1, MSCState1} = - lists:foldl(fun ({MsgId, _SeqId, _MsgProps, IsPersistent, + lists:foldl(fun ({MsgId, _SeqId, MsgProps, IsPersistent, _IsDelivered}, {Acc0, MSCState0}) -> {{ok, Msg = #basic_message {}}, MSCState1} = msg_store_read(MSCState0, IsPersistent, MsgId), - {Fun(Msg, Acc0), MSCState1} + {Fun(Msg, MsgProps, Acc0), MSCState1} end, {Acc, MSCState}, List), delta_fold(Fun, Acc1, DeltaSeqId1, DeltaSeqIdEnd, State #vqstate { index_state = IndexState1, @@ -1532,8 +1528,7 @@ limit_ram_acks(Quota, State = #vqstate { pending_ack = PA, {Quota, State}; false -> {SeqId, RAI1} = gb_sets:take_largest(RAI), - MsgStatus = #msg_status { is_persistent = false} = - gb_trees:get(SeqId, PA), + MsgStatus = gb_trees:get(SeqId, PA), {MsgStatus1, State1} = maybe_write_to_disk(true, false, MsgStatus, State), PA1 = gb_trees:update(SeqId, m(trim_msg_status(MsgStatus1)), PA), |