diff options
Diffstat (limited to 'src/rabbit_invariable_queue.erl')
| -rw-r--r-- | src/rabbit_invariable_queue.erl | 314 |
1 files changed, 0 insertions, 314 deletions
diff --git a/src/rabbit_invariable_queue.erl b/src/rabbit_invariable_queue.erl deleted file mode 100644 index 5a0532ea..00000000 --- a/src/rabbit_invariable_queue.erl +++ /dev/null @@ -1,314 +0,0 @@ -%% The contents of this file are subject to the Mozilla Public License -%% Version 1.1 (the "License"); you may not use this file except in -%% compliance with the License. You may obtain a copy of the License at -%% http://www.mozilla.org/MPL/ -%% -%% Software distributed under the License is distributed on an "AS IS" -%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the -%% License for the specific language governing rights and limitations -%% under the License. -%% -%% The Original Code is RabbitMQ. -%% -%% The Initial Developers of the Original Code are LShift Ltd, -%% Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd. -%% -%% Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, -%% Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd -%% are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial -%% Technologies LLC, and Rabbit Technologies Ltd. -%% -%% Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift -%% Ltd. Portions created by Cohesive Financial Technologies LLC are -%% Copyright (C) 2007-2010 Cohesive Financial Technologies -%% LLC. Portions created by Rabbit Technologies Ltd are Copyright -%% (C) 2007-2010 Rabbit Technologies Ltd. -%% -%% All Rights Reserved. -%% -%% Contributor(s): ______________________________________. -%% - --module(rabbit_invariable_queue). - --export([init/3, terminate/1, delete_and_terminate/1, purge/1, publish/3, - publish_delivered/4, fetch/2, ack/2, tx_publish/4, tx_ack/3, - dropwhile/2, tx_rollback/2, tx_commit/4, requeue/3, len/1, is_empty/1, - set_ram_duration_target/2, ram_duration/1, needs_idle_timeout/1, - idle_timeout/1, handle_pre_hibernate/1, status/1]). - --export([start/1, stop/0]). - --behaviour(rabbit_backing_queue). - --include("rabbit.hrl"). - --record(iv_state, { queue, qname, durable, len, pending_ack }). --record(tx, { pending_messages, pending_acks, is_persistent }). - --ifdef(use_specs). - --type(ack() :: rabbit_guid:guid() | 'blank_ack'). --type(state() :: #iv_state { queue :: queue(), - qname :: rabbit_amqqueue:name(), - len :: non_neg_integer(), - pending_ack :: dict() - }). --include("rabbit_backing_queue_spec.hrl"). - --endif. - -start(DurableQueues) -> - ok = rabbit_sup:start_child(rabbit_persister, [DurableQueues]). - -stop() -> - ok = rabbit_sup:stop_child(rabbit_persister). - -init(QName, IsDurable, Recover) -> - Q = queue:from_list(case IsDurable andalso Recover of - true -> rabbit_persister:queue_content(QName); - false -> [] - end), - #iv_state { queue = Q, - qname = QName, - durable = IsDurable, - len = queue:len(Q), - pending_ack = dict:new() }. - -terminate(State) -> - State #iv_state { queue = queue:new(), len = 0, pending_ack = dict:new() }. - -delete_and_terminate(State = #iv_state { qname = QName, durable = IsDurable, - pending_ack = PA }) -> - ok = persist_acks(QName, IsDurable, none, dict:fetch_keys(PA), PA), - {_PLen, State1} = purge(State), - terminate(State1). - -purge(State = #iv_state { queue = Q, qname = QName, durable = IsDurable, - len = Len }) -> - %% We do not purge messages pending acks. - {AckTags, PA} = - rabbit_misc:queue_fold( - fun ({#basic_message { is_persistent = false }, - _MsgProps, _IsDelivered}, Acc) -> - Acc; - ({Msg = #basic_message { guid = Guid }, MsgProps, IsDelivered}, - {AckTagsN, PAN}) -> - ok = persist_delivery(QName, IsDurable, IsDelivered, Msg), - {[Guid | AckTagsN], store_ack(Msg, MsgProps, PAN)} - end, {[], dict:new()}, Q), - ok = persist_acks(QName, IsDurable, none, AckTags, PA), - {Len, State #iv_state { len = 0, queue = queue:new() }}. - -publish(Msg, MsgProps, State = #iv_state { queue = Q, - qname = QName, - durable = IsDurable, - len = Len }) -> - ok = persist_message(QName, IsDurable, none, Msg, MsgProps), - State #iv_state { queue = enqueue(Msg, MsgProps, false, Q), len = Len + 1 }. - -publish_delivered(false, _Msg, _MsgProps, State) -> - {blank_ack, State}; -publish_delivered(true, Msg = #basic_message { guid = Guid }, - MsgProps, - State = #iv_state { qname = QName, durable = IsDurable, - len = 0, pending_ack = PA }) -> - ok = persist_message(QName, IsDurable, none, Msg, MsgProps), - ok = persist_delivery(QName, IsDurable, false, Msg), - {Guid, State #iv_state { pending_ack = store_ack(Msg, MsgProps, PA) }}. - -dropwhile(_Pred, State = #iv_state { len = 0 }) -> - State; -dropwhile(Pred, State = #iv_state { queue = Q }) -> - {{value, {Msg, MsgProps, IsDelivered}}, Q1} = queue:out(Q), - case Pred(MsgProps) of - true -> {_, State1} = fetch_internal(false, Q1, Msg, MsgProps, - IsDelivered, State), - dropwhile(Pred, State1); - false -> State - end. - -fetch(_AckRequired, State = #iv_state { len = 0 }) -> - {empty, State}; -fetch(AckRequired, State = #iv_state { queue = Q }) -> - {{value, {Msg, MsgProps, IsDelivered}}, Q1} = queue:out(Q), - fetch_internal(AckRequired, Q1, Msg, MsgProps, IsDelivered, State). - -fetch_internal(AckRequired, Q1, - Msg = #basic_message { guid = Guid }, - MsgProps, IsDelivered, - State = #iv_state { len = Len, - qname = QName, - durable = IsDurable, - pending_ack = PA }) -> - Len1 = Len - 1, - ok = persist_delivery(QName, IsDurable, IsDelivered, Msg), - PA1 = store_ack(Msg, MsgProps, PA), - {AckTag, PA2} = case AckRequired of - true -> {Guid, PA1}; - false -> ok = persist_acks(QName, IsDurable, none, - [Guid], PA1), - {blank_ack, PA} - end, - {{Msg, IsDelivered, AckTag, Len1}, - State #iv_state { queue = Q1, len = Len1, pending_ack = PA2 }}. - -ack(AckTags, State = #iv_state { qname = QName, durable = IsDurable, - pending_ack = PA }) -> - ok = persist_acks(QName, IsDurable, none, AckTags, PA), - PA1 = remove_acks(AckTags, PA), - State #iv_state { pending_ack = PA1 }. - -tx_publish(Txn, Msg, MsgProps, State = #iv_state { qname = QName, - durable = IsDurable }) -> - Tx = #tx { pending_messages = Pubs } = lookup_tx(Txn), - store_tx(Txn, Tx #tx { pending_messages = [{Msg, MsgProps} | Pubs] }), - ok = persist_message(QName, IsDurable, Txn, Msg, MsgProps), - State. - -tx_ack(Txn, AckTags, State = #iv_state { qname = QName, durable = IsDurable, - pending_ack = PA }) -> - Tx = #tx { pending_acks = Acks } = lookup_tx(Txn), - store_tx(Txn, Tx #tx { pending_acks = [AckTags | Acks] }), - ok = persist_acks(QName, IsDurable, Txn, AckTags, PA), - State. - -tx_rollback(Txn, State = #iv_state { qname = QName }) -> - #tx { pending_acks = AckTags } = lookup_tx(Txn), - ok = do_if_persistent(fun rabbit_persister:rollback_transaction/1, - Txn, QName), - erase_tx(Txn), - {lists:flatten(AckTags), State}. - -tx_commit(Txn, Fun, MsgPropsFun, State = #iv_state { qname = QName, - pending_ack = PA, - queue = Q, - len = Len }) -> - #tx { pending_acks = AckTags, pending_messages = PubsRev } = lookup_tx(Txn), - ok = do_if_persistent(fun rabbit_persister:commit_transaction/1, - Txn, QName), - erase_tx(Txn), - Fun(), - AckTags1 = lists:flatten(AckTags), - PA1 = remove_acks(AckTags1, PA), - {Q1, Len1} = lists:foldr(fun ({Msg, MsgProps}, {QN, LenN}) -> - {enqueue(Msg, MsgPropsFun(MsgProps), - false, QN), - LenN + 1} - end, {Q, Len}, PubsRev), - {AckTags1, State #iv_state { pending_ack = PA1, queue = Q1, len = Len1 }}. - -requeue(AckTags, MsgPropsFun, State = #iv_state { pending_ack = PA, - queue = Q, - len = Len }) -> - %% We don't need to touch the persister here - the persister will - %% already have these messages published and delivered as - %% necessary. The complication is that the persister's seq_id will - %% now be wrong, given the position of these messages in our queue - %% here. However, the persister's seq_id is only used for sorting - %% on startup, and requeue is silent as to where the requeued - %% messages should appear, thus the persister is permitted to sort - %% based on seq_id, even though it'll likely give a different - %% order to the last known state of our queue, prior to shutdown. - {Q1, Len1} = lists:foldl( - fun (Guid, {QN, LenN}) -> - {Msg = #basic_message {}, MsgProps} - = dict:fetch(Guid, PA), - {enqueue(Msg, MsgPropsFun(MsgProps), true, QN), - LenN + 1} - end, {Q, Len}, AckTags), - PA1 = remove_acks(AckTags, PA), - State #iv_state { pending_ack = PA1, queue = Q1, len = Len1 }. - -enqueue(Msg, MsgProps, IsDelivered, Q) -> - queue:in({Msg, MsgProps, IsDelivered}, Q). - -len(#iv_state { len = Len }) -> Len. - -is_empty(State) -> 0 == len(State). - -set_ram_duration_target(_DurationTarget, State) -> State. - -ram_duration(State) -> {0, State}. - -needs_idle_timeout(_State) -> false. - -idle_timeout(State) -> State. - -handle_pre_hibernate(State) -> State. - -status(_State) -> []. - -%%---------------------------------------------------------------------------- - -remove_acks(AckTags, PA) -> lists:foldl(fun dict:erase/2, PA, AckTags). - -store_ack(Msg = #basic_message { guid = Guid }, MsgProps, PA) -> - dict:store(Guid, {Msg, MsgProps}, PA). - -%%---------------------------------------------------------------------------- - -lookup_tx(Txn) -> - case get({txn, Txn}) of - undefined -> #tx { pending_messages = [], - pending_acks = [], - is_persistent = false }; - V -> V - end. - -store_tx(Txn, Tx) -> - put({txn, Txn}, Tx). - -erase_tx(Txn) -> - erase({txn, Txn}). - -mark_tx_persistent(Txn) -> - store_tx(Txn, (lookup_tx(Txn)) #tx { is_persistent = true }). - -is_tx_persistent(Txn) -> - (lookup_tx(Txn)) #tx.is_persistent. - -do_if_persistent(F, Txn, QName) -> - ok = case is_tx_persistent(Txn) of - false -> ok; - true -> F({Txn, QName}) - end. - -%%---------------------------------------------------------------------------- - -persist_message(QName, true, Txn, Msg = #basic_message { - is_persistent = true }, MsgProps) -> - Msg1 = Msg #basic_message { - %% don't persist any recoverable decoded properties - content = rabbit_binary_parser:clear_decoded_content( - Msg #basic_message.content)}, - persist_work(Txn, QName, - [{publish, Msg1, MsgProps, - {QName, Msg1 #basic_message.guid}}]); -persist_message(_QName, _IsDurable, _Txn, _Msg, _MsgProps) -> - ok. - -persist_delivery(QName, true, false, #basic_message { is_persistent = true, - guid = Guid }) -> - persist_work(none, QName, [{deliver, {QName, Guid}}]); -persist_delivery(_QName, _IsDurable, _IsDelivered, _Msg) -> - ok. - -persist_acks(QName, true, Txn, AckTags, PA) -> - persist_work(Txn, QName, - [{ack, {QName, Guid}} || Guid <- AckTags, - begin - {Msg, _MsgProps} - = dict:fetch(Guid, PA), - Msg #basic_message.is_persistent - end]); -persist_acks(_QName, _IsDurable, _Txn, _AckTags, _PA) -> - ok. - -persist_work(_Txn,_QName, []) -> - ok; -persist_work(none, _QName, WorkList) -> - rabbit_persister:dirty_work(WorkList); -persist_work(Txn, QName, WorkList) -> - mark_tx_persistent(Txn), - rabbit_persister:extend_transaction({Txn, QName}, WorkList). |