diff options
-rw-r--r-- | src/rabbit_amqqueue_process.erl | 4 | ||||
-rw-r--r-- | src/rabbit_backing_queue.erl | 12 | ||||
-rw-r--r-- | src/rabbit_mirror_queue_master.erl | 18 |
3 files changed, 14 insertions, 20 deletions
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index ae883852..11079523 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -548,9 +548,7 @@ attempt_delivery(Delivery = #delivery{sender = SenderPid, message = Message}, {{Message, Delivered, undefined}, true, discard(Delivery, State1)} end, false, State#q{backing_queue_state = BQS1}); - {published, BQS1} -> - {true, State#q{backing_queue_state = BQS1}}; - {discarded, BQS1} -> + {true, BQS1} -> {true, State#q{backing_queue_state = BQS1}} end. diff --git a/src/rabbit_backing_queue.erl b/src/rabbit_backing_queue.erl index 2f247448..bf26cb5a 100644 --- a/src/rabbit_backing_queue.erl +++ b/src/rabbit_backing_queue.erl @@ -90,10 +90,7 @@ -> {ack(), state()}. %% Called to inform the BQ about messages which have reached the -%% queue, but are not going to be further passed to BQ for some -%% reason. Note that this may be invoked for messages for which -%% BQ:is_duplicate/2 has already returned {'published' | 'discarded', -%% BQS}. +%% queue, but are not going to be further passed to BQ. -callback discard(rabbit_types:msg_id(), pid(), state()) -> state(). %% Return ids of messages which have been confirmed since the last @@ -216,11 +213,10 @@ -callback invoke(atom(), fun ((atom(), A) -> A), state()) -> state(). %% Called prior to a publish or publish_delivered call. Allows the BQ -%% to signal that it's already seen this message (and in what capacity -%% - i.e. was it published previously or discarded previously) and -%% thus the message should be dropped. +%% to signal that it's already seen this message, (e.g. it was published +%% or discarded previously) and thus the message should be dropped. -callback is_duplicate(rabbit_types:basic_message(), state()) - -> {'false'|'published'|'discarded', state()}. + -> {boolean(), state()}. -else. diff --git a/src/rabbit_mirror_queue_master.erl b/src/rabbit_mirror_queue_master.erl index 3b49a6b8..572cd0ca 100644 --- a/src/rabbit_mirror_queue_master.erl +++ b/src/rabbit_mirror_queue_master.erl @@ -382,8 +382,9 @@ is_duplicate(Message = #basic_message { id = MsgId }, %% immediately after calling is_duplicate). The msg is %% invalid. We will not see this again, nor will we be %% further involved in confirming this message, so erase. - {published, State #state { seen_status = dict:erase(MsgId, SS) }}; - {ok, confirmed} -> + {true, State #state { seen_status = dict:erase(MsgId, SS) }}; + {ok, Disposition} + when Disposition =:= confirmed %% It got published when we were a slave via gm, and %% confirmed some time after that (maybe even after %% promotion), but before we received the publish from the @@ -392,13 +393,12 @@ is_duplicate(Message = #basic_message { id = MsgId }, %% need to confirm now. As above, amqqueue_process will %% have the entry for the msg_id_to_channel mapping added %% immediately after calling is_duplicate/2. - {published, State #state { seen_status = dict:erase(MsgId, SS), - confirmed = [MsgId | Confirmed] }}; - {ok, discarded} -> - %% Message was discarded while we were a slave. - %% Erase and confirm. - {discarded, State #state { seen_status = dict:erase(MsgId, SS), - confirmed = [MsgId | Confirmed] }} + orelse Disposition =:= discarded -> + %% Message was discarded while we were a slave. Confirm now. + %% As above, amqqueue_process will have the entry for the + %% msg_id_to_channel mapping. + {true, State #state { seen_status = dict:erase(MsgId, SS), + confirmed = [MsgId | Confirmed] }} end. %% --------------------------------------------------------------------------- |