diff --git a/src/rmq/rmqamqp/rmqamqp_receivechannel.cpp b/src/rmq/rmqamqp/rmqamqp_receivechannel.cpp index c1bbe949..1b2a8680 100644 --- a/src/rmq/rmqamqp/rmqamqp_receivechannel.cpp +++ b/src/rmq/rmqamqp/rmqamqp_receivechannel.cpp @@ -608,6 +608,12 @@ void ReceiveChannel::consumeAckBatchFromQueue() bsl::shared_ptr batch = d_ackQueue->popFront(); bsl::vector& acks = batch->lock(); + if (state() == CHANNEL_CLOSE_SENT || state() == CLOSED) { + BALL_LOG_WARN << "Ignoring " << acks.size() + << " ack/nacks for a channel which is closing."; + return; + } + bsl::vector::iterator valid = acks.begin(); for (bsl::vector::iterator it = valid; it < acks.end(); ++it) { diff --git a/src/tests/rmqamqp/rmqamqp_receivechannel.t.cpp b/src/tests/rmqamqp/rmqamqp_receivechannel.t.cpp index 5b602312..268256ed 100644 --- a/src/tests/rmqamqp/rmqamqp_receivechannel.t.cpp +++ b/src/tests/rmqamqp/rmqamqp_receivechannel.t.cpp @@ -710,6 +710,60 @@ TEST_F(ReceiveChannelTests, ServerCancels) EXPECT_THAT(receiveChannel->state(), Eq(rmqamqp::Channel::CLOSED)); } +TEST_F(ReceiveChannelTests, NoAckAfterChannelCloseSent) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupConsumer(*receiveChannel, "consumer1"); + + const rmqt::Envelope envelope(1, + receiveChannel->lifetimeId(), + "consumer1", + "exchange", + "routing-key", + false); + receiveMessage(*receiveChannel, envelope.deliveryTag(), "consumer1"); + + EXPECT_CALL(d_callback, onAsyncWrite(_, _)).WillOnce(InvokeArgument<1>()); + receiveChannel->processReceived(rmqamqp::Message(rmqamqpt::Method( + rmqamqpt::BasicMethod(rmqamqpt::BasicCancel("consumer1"))))); + ASSERT_THAT(receiveChannel->state(), + Eq(rmqamqp::Channel::CHANNEL_CLOSE_SENT)); + + EXPECT_CALL(d_callback, onAsyncWrite(_, _)).Times(0); + ackMessage(*receiveChannel, envelope); +} + +TEST_F(ReceiveChannelTests, NoAckAfterChannelCloseOkSent) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupConsumer(*receiveChannel, "consumer1"); + + const rmqt::Envelope envelope(1, + receiveChannel->lifetimeId(), + "consumer1", + "exchange", + "routing-key", + false); + receiveMessage(*receiveChannel, envelope.deliveryTag(), "consumer1"); + + EXPECT_CALL( + d_callback, + onAsyncWrite(Pointee(MethodMsgTypeEq(rmqamqpt::Method( + rmqamqpt::ChannelMethod(rmqamqpt::ChannelCloseOk())))), + _)); + receiveChannel->processReceived(rmqamqp::Message( + rmqamqpt::Method(rmqamqpt::ChannelMethod(rmqamqpt::ChannelClose( + rmqamqpt::Constants::REPLY_SUCCESS, "Closing Channel"))))); + ASSERT_THAT(receiveChannel->state(), Eq(rmqamqp::Channel::CLOSED)); + + EXPECT_CALL(d_callback, onAsyncWrite(_, _)).Times(0); + ackMessage(*receiveChannel, envelope); +} + TEST_F(ReceiveChannelTests, AckMessage) { bsl::shared_ptr receiveChannel = makeReceiveChannel(1);