diff options
| author | Rob Harrop <rob@rabbitmq.com> | 2011-01-11 17:06:38 +0000 |
|---|---|---|
| committer | Rob Harrop <rob@rabbitmq.com> | 2011-01-11 17:06:38 +0000 |
| commit | 2be3dddb891f6edec591e8e69742b30063719d55 (patch) | |
| tree | eb6795122d535e570fbed92ae70d25114d05b2ff /src | |
| parent | 79acba27de543f92ef9bf05034671372ac15b3f4 (diff) | |
| parent | 32be493dc092069bc496fa006d16f942d953d0a3 (diff) | |
| download | rabbitmq-server-git-2be3dddb891f6edec591e8e69742b30063719d55.tar.gz | |
Merge with default
Diffstat (limited to 'src')
| -rw-r--r-- | src/rabbit_channel.erl | 24 |
1 files changed, 16 insertions, 8 deletions
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl index 930e48e6cd..1ad92318f8 100644 --- a/src/rabbit_channel.erl +++ b/src/rabbit_channel.erl @@ -568,6 +568,11 @@ handle_method(#'basic.publish'{exchange = ExchangeNameBin, _ -> add_tx_participants(DeliveredQPids, State2) end}; +handle_method(#'basic.nack'{delivery_tag = DeliveryTag, + multiple = Multiple}, + _, State) -> + reject(DeliveryTag, true, Multiple, State); + handle_method(#'basic.ack'{delivery_tag = DeliveryTag, multiple = Multiple}, _, State = #ch{transaction_id = TxnKey, @@ -753,14 +758,8 @@ handle_method(#'basic.recover'{requeue = Requeue}, Content, State) -> handle_method(#'basic.reject'{delivery_tag = DeliveryTag, requeue = Requeue}, - _, State = #ch{unacked_message_q = UAMQ}) -> - {Acked, Remaining} = collect_acks(UAMQ, DeliveryTag, false), - ok = fold_per_queue( - fun (QPid, MsgIds, ok) -> - rabbit_amqqueue:reject(QPid, MsgIds, Requeue, self()) - end, ok, Acked), - ok = notify_limiter(State#ch.limiter_pid, Acked), - {noreply, State#ch{unacked_message_q = Remaining}}; + _, State) -> + reject(DeliveryTag, Requeue, false, State); handle_method(#'exchange.declare'{exchange = ExchangeNameBin, type = TypeNameBin, @@ -1078,6 +1077,15 @@ basic_return(#basic_message{exchange_name = ExchangeName, routing_key = RoutingKey}, Content). +reject(DeliveryTag, Requeue, Multiple, State = #ch{unacked_message_q = UAMQ}) -> + {Acked, Remaining} = collect_acks(UAMQ, DeliveryTag, Multiple), + ok = fold_per_queue( + fun (QPid, MsgIds, ok) -> + rabbit_amqqueue:reject(QPid, MsgIds, Requeue, self()) + end, ok, Acked), + ok = notify_limiter(State#ch.limiter_pid, Acked), + {noreply, State#ch{unacked_message_q = Remaining}}. + ack_record(DeliveryTag, ConsumerTag, _MsgStruct = {_QName, QPid, MsgId, _Redelivered, _Msg}) -> {DeliveryTag, ConsumerTag, {QPid, MsgId}}. |