diff --git a/src/rmq/rmqamqp/rmqamqp_channel.cpp b/src/rmq/rmqamqp/rmqamqp_channel.cpp index 2c847a6..4040010 100644 --- a/src/rmq/rmqamqp/rmqamqp_channel.cpp +++ b/src/rmq/rmqamqp/rmqamqp_channel.cpp @@ -462,7 +462,7 @@ Channel::CleanupIndicator Channel::reset(bool startRetry) d_metricPublisher->publishCounter( "channel_disconnected", 1, getVHostAndChannelTags()); - return KEEP; + return d_permanentlyClosing ? CLEANUP : KEEP; } void Channel::retry(const bsl::weak_ptr& weakSelf) @@ -498,6 +498,11 @@ void Channel::ready() void Channel::open() { + if (d_permanentlyClosing) { + BALL_LOG_DEBUG << "Not opening a channel which is permanently closing"; + return; + } + using bdlf::PlaceHolders::_1; d_hungProgressTimer->start( bdlf::BindUtil::bind(&Channel::channelHung, weak_from_this(), _1)); @@ -526,6 +531,11 @@ void Channel::writeMessage( void Channel::gracefulClose() { d_permanentlyClosing = true; + + if (d_state == CLOSED || d_state == CHANNEL_CLOSE_SENT) { + return; + } + close(rmqamqpt::Constants::REPLY_SUCCESS, "Closing Channel"); } diff --git a/src/tests/rmqamqp/rmqamqp_channel.t.cpp b/src/tests/rmqamqp/rmqamqp_channel.t.cpp index b1eede6..70f217e 100644 --- a/src/tests/rmqamqp/rmqamqp_channel.t.cpp +++ b/src/tests/rmqamqp/rmqamqp_channel.t.cpp @@ -196,19 +196,41 @@ TEST_F(ChannelTests, OpenChannel) bsl::shared_ptr channel = makeOpeningChannel(d_topology); } -TEST_F(ChannelTests, GracefulCloseChannel) +TEST_F(ChannelTests, GracefulCloseOnClosedChannelNeverReopens) { bsl::shared_ptr mockChannel = makeChannel(d_topology); - rmqamqpt::ChannelClose closeMethod(rmqamqpt::Constants::REPLY_SUCCESS, - "Closing Channel"); - EXPECT_CALL(d_callback, - onAsyncWrite(::testing::Pointee( - MessageEq(rmqamqp::Message(rmqamqpt::Method( - rmqamqpt::ChannelMethod(closeMethod))))), - _)); + EXPECT_CALL(d_callback, onAsyncWrite(_, _)).Times(0); + + mockChannel->gracefulClose(); + mockChannel->open(); +} + +TEST_F(ChannelTests, GracefulCloseWhileClosingCleansUpOnCloseOk) +{ + bsl::shared_ptr mockChannel = makeChannel(d_topology); + mockChannel->setState(rmqamqp::Channel::CHANNEL_CLOSE_SENT); + + EXPECT_CALL(d_callback, onAsyncWrite(_, _)).Times(0); mockChannel->gracefulClose(); + + rmqamqpt::ChannelCloseOk closeOkMethod; + EXPECT_THAT(mockChannel->processReceived(rmqamqp::Message( + rmqamqpt::Method(rmqamqpt::ChannelMethod(closeOkMethod)))), + Eq(rmqamqp::Channel::CLEANUP)); +} + +TEST_F(ChannelTests, ResetDuringGracefulCloseReturnsCleanup) +{ + bsl::shared_ptr mockChannel = + makeOpeningChannel(d_topology); + mockChannel->setState(rmqamqp::Channel::READY); + + EXPECT_CALL(d_callback, onAsyncWrite(_, _)); + mockChannel->gracefulClose(); + + EXPECT_THAT(mockChannel->reset(), Eq(rmqamqp::Channel::CLEANUP)); } TEST_F(ChannelTests, closingProcessCloseOkReturnsCleanup)