diff --git a/src/rmq/rmqamqp/rmqamqp_channel.cpp b/src/rmq/rmqamqp/rmqamqp_channel.cpp index 76d18cd5..34a98a18 100644 --- a/src/rmq/rmqamqp/rmqamqp_channel.cpp +++ b/src/rmq/rmqamqp/rmqamqp_channel.cpp @@ -145,9 +145,19 @@ class Channel::ChannelMethodProcessor { BALL_LOG_INFO << "Received ChannelClose from server: " << closeMethod; rmqamqpt::ChannelCloseOk closeOkMethod; - d_channel.writeMessage( - Message(rmqamqpt::Method(rmqamqpt::ChannelMethod(closeOkMethod))), - CLOSED); + const Message closeOk = + Message(rmqamqpt::Method(rmqamqpt::ChannelMethod(closeOkMethod))); + + if (d_channel.d_state == CHANNEL_CLOSE_SENT) { + d_channel.writeMessage(closeOk, &noopWriteHandler); + } + else { + d_channel.updateState(CLOSED); + d_channel.writeMessage( + closeOk, + bdlf::BindUtil::bind(&Channel::onCloseOkWritten, + d_channel.weak_from_this())); + } if (closeMethod.classId() || closeMethod.methodId()) { (d_channel.d_retryHandler->errorCallback())( @@ -160,6 +170,12 @@ class Channel::ChannelMethodProcessor { ResultType operator()(const rmqamqpt::ChannelCloseOk&) const { + if (d_channel.d_state == CHANNEL_OPEN_SENT) { + BALL_LOG_INFO << "Ignoring channel CLOSE-OK from before the " + "channel was reopened"; + return KEEP; + } + if (d_channel.d_state != CHANNEL_CLOSE_SENT) { BALL_LOG_ERROR << "Received channel CLOSE-OK method from server " "without sending CLOSE method"; @@ -443,19 +459,16 @@ void Channel::processMessage(const rmqt::Message&) BALL_LOG_ERROR << "Throwing Content Away"; } -void Channel::onWriteComplete(const bsl::weak_ptr& weakSelf) +void Channel::onCloseOkWritten(const bsl::weak_ptr& weakSelf) { bsl::shared_ptr self = weakSelf.lock(); if (!self) { - BALL_LOG_TRACE - << "Write completion handler called after channel destroyed"; + BALL_LOG_TRACE << "Close-Ok write completed after channel destroyed"; return; } - if (self->d_state == CLOSED) { - const bool startRetry = !self->d_permanentlyClosing; - self->reset(startRetry); - } + const bool startRetry = !self->d_permanentlyClosing; + self->reset(startRetry); } Channel::CleanupIndicator Channel::reset(bool startRetry) @@ -466,8 +479,8 @@ Channel::CleanupIndicator Channel::reset(bool startRetry) onReset(); if (startRetry) { - d_retryHandler->retry( - bdlf::BindUtil::bind(&Channel::retry, weak_from_this())); + d_retryHandler->retry(bdlf::BindUtil::bind( + &Channel::retry, weak_from_this(), lifetimeId())); } d_metricPublisher->publishCounter( "channel_disconnected", 1, getVHostAndChannelTags()); @@ -475,7 +488,8 @@ Channel::CleanupIndicator Channel::reset(bool startRetry) return d_permanentlyClosing ? CLEANUP : KEEP; } -void Channel::retry(const bsl::weak_ptr& weakSelf) +void Channel::retry(const bsl::weak_ptr& weakSelf, + size_t scheduledLifetimeId) { bsl::shared_ptr self = weakSelf.lock(); if (!self) { @@ -483,6 +497,13 @@ void Channel::retry(const bsl::weak_ptr& weakSelf) return; } + if (self->lifetimeId() != scheduledLifetimeId) { + BALL_LOG_INFO << "Ignoring retry scheduled in channel lifetime " + << scheduledLifetimeId << ", current lifetime is " + << self->lifetimeId(); + return; + } + self->open(); } @@ -513,6 +534,11 @@ void Channel::open() return; } + if (d_state != CLOSED) { + BALL_LOG_INFO << "Not opening a channel in state " << d_state; + return; + } + using bdlf::PlaceHolders::_1; d_hungProgressTimer->start( bdlf::BindUtil::bind(&Channel::channelHung, weak_from_this(), _1)); @@ -526,9 +552,7 @@ void Channel::writeMessage(const rmqamqp::Message& message, State newState) { updateState(newState); BALL_LOG_TRACE << "State now set to: " << d_state; - d_onAsyncWrite( - bsl::make_shared(message), - bdlf::BindUtil::bind(&Channel::onWriteComplete, weak_from_this())); + d_onAsyncWrite(bsl::make_shared(message), &noopWriteHandler); } void Channel::writeMessage( diff --git a/src/rmq/rmqamqp/rmqamqp_channel.h b/src/rmq/rmqamqp/rmqamqp_channel.h index 663ba7cd..3caeacd6 100644 --- a/src/rmq/rmqamqp/rmqamqp_channel.h +++ b/src/rmq/rmqamqp/rmqamqp_channel.h @@ -152,7 +152,7 @@ class Channel : public bsl::enable_shared_from_this { void ready(); // Callback for writes - static void onWriteComplete(const bsl::weak_ptr& weakSelf); + static void onCloseOkWritten(const bsl::weak_ptr& weakSelf); void writeMessage(const rmqamqp::Message& message, State newState); @@ -160,7 +160,8 @@ class Channel : public bsl::enable_shared_from_this { const rmqio::Connection::SuccessWriteCallback& callBack); /// Callback for re-opening channel - static void retry(const bsl::weak_ptr& weakSelf); + static void retry(const bsl::weak_ptr& weakSelf, + size_t scheduledLifetimeId); const bsl::string& vhostName() const; diff --git a/src/tests/rmqamqp/rmqamqp_channel.t.cpp b/src/tests/rmqamqp/rmqamqp_channel.t.cpp index c671a589..7f304cac 100644 --- a/src/tests/rmqamqp/rmqamqp_channel.t.cpp +++ b/src/tests/rmqamqp/rmqamqp_channel.t.cpp @@ -199,6 +199,24 @@ TEST_F(ChannelTests, OpenChannel) bsl::shared_ptr channel = makeOpeningChannel(d_topology); } +TEST_F(ChannelTests, OpenWhileOpeningSendsNothing) +{ + bsl::shared_ptr channel = makeOpeningChannel(d_topology); + + EXPECT_CALL(d_callback, onAsyncWrite(_, _)).Times(0); + channel->open(); +} + +TEST_F(ChannelTests, CloseOkWhileOpeningSendsNothing) +{ + bsl::shared_ptr channel = makeOpeningChannel(d_topology); + + EXPECT_CALL(*d_retryHandler, retry(_)).WillRepeatedly(InvokeArgument<0>()); + EXPECT_CALL(d_callback, onAsyncWrite(_, _)).Times(0); + channel->processReceived(rmqamqp::Message( + rmqamqpt::Method(rmqamqpt::ChannelMethod(rmqamqpt::ChannelCloseOk())))); +} + TEST_F(ChannelTests, GracefulCloseOnClosedChannelNeverReopens) { bsl::shared_ptr mockChannel = makeChannel(d_topology); @@ -306,6 +324,73 @@ TEST_F(ChannelTests, ReceiveCloseMethodReOpens) EXPECT_THAT(mockChannel->state(), Eq(rmqamqp::Channel::CHANNEL_OPEN_SENT)); } +TEST_F(ChannelTests, ServerCloseRetriesOnceWithWritesInFlight) +{ + bsl::shared_ptr mockChannel = + makeOpeningChannel(d_topology); + + bsl::function queueDeclareWritten; + EXPECT_CALL(d_callback, onAsyncWrite(_, _)) + .WillOnce(SaveArg<1>(&queueDeclareWritten)); + mockChannel->processReceived(rmqamqp::Message( + rmqamqpt::Method(rmqamqpt::ChannelMethod(rmqamqpt::ChannelOpenOk())))); + + bsl::function closeOkWritten; + EXPECT_CALL(d_callback, onAsyncWrite(_, _)) + .WillOnce(SaveArg<1>(&closeOkWritten)); + mockChannel->processReceived(rmqamqp::Message( + rmqamqpt::Method(rmqamqpt::ChannelMethod(rmqamqpt::ChannelClose( + rmqamqpt::Constants::REPLY_SUCCESS, "Closing Channel"))))); + + EXPECT_CALL(*d_retryHandler, retry(_)); + queueDeclareWritten(); + closeOkWritten(); +} + +TEST_F(ChannelTests, ServerCloseCrossingGracefulCloseCleansUpOnCloseOk) +{ + bsl::shared_ptr mockChannel = + makeOpeningChannel(d_topology); + mockChannel->setState(rmqamqp::Channel::READY); + + EXPECT_CALL(d_callback, onAsyncWrite(_, _)).WillOnce(InvokeArgument<1>()); + mockChannel->gracefulClose(); + + EXPECT_CALL(d_callback, + onAsyncWrite( + Pointee(MessageEq(rmqamqp::Message(rmqamqpt::Method( + rmqamqpt::ChannelMethod(rmqamqpt::ChannelCloseOk()))))), + _)) + .WillOnce(InvokeArgument<1>()); + mockChannel->processReceived(rmqamqp::Message( + rmqamqpt::Method(rmqamqpt::ChannelMethod(rmqamqpt::ChannelClose( + rmqamqpt::Constants::REPLY_SUCCESS, "Closing Channel"))))); + + EXPECT_THAT(mockChannel->processReceived(rmqamqp::Message(rmqamqpt::Method( + rmqamqpt::ChannelMethod(rmqamqpt::ChannelCloseOk())))), + Eq(rmqamqp::Channel::CLEANUP)); +} + +TEST_F(ChannelTests, ServerCloseCrossingOurCloseReopensOnlyAfterCloseOk) +{ + bsl::shared_ptr mockChannel = + makeOpeningChannel(d_topology); + mockChannel->setState(rmqamqp::Channel::READY); + + EXPECT_CALL(d_callback, onAsyncWrite(_, _)).WillOnce(InvokeArgument<1>()); + mockChannel->close(rmqamqpt::Constants::CHANNEL_ERROR, "Channel error"); + + EXPECT_CALL(*d_retryHandler, retry(_)).Times(0); + EXPECT_CALL(d_callback, onAsyncWrite(_, _)).WillOnce(InvokeArgument<1>()); + mockChannel->processReceived(rmqamqp::Message( + rmqamqpt::Method(rmqamqpt::ChannelMethod(rmqamqpt::ChannelClose( + rmqamqpt::Constants::REPLY_SUCCESS, "Closing Channel"))))); + + EXPECT_CALL(*d_retryHandler, retry(_)); + mockChannel->processReceived(rmqamqp::Message( + rmqamqpt::Method(rmqamqpt::ChannelMethod(rmqamqpt::ChannelCloseOk())))); +} + TEST_F(ChannelTests, ReceiveOpenOkMethod) { rmqt::Topology emptyTopology; @@ -973,19 +1058,22 @@ TEST_F(ChannelTests, sendBindAndUnbind) EXPECT_TRUE(future2.tryResult()); } -TEST_F(ChannelTests, onWriteCompleteCbValidity) +TEST_F(ChannelTests, onCloseOkWrittenCbValidity) { - // Ensure that the onWriteComplete callback can be invoked after the Channel - // is destructed. + // Ensure that the onCloseOkWritten callback can be invoked after the + // Channel is destructed. bsl::function callbackFn; { - bsl::shared_ptr channel = makeChannel(d_topology); + bsl::shared_ptr channel = + makeOpeningChannel(d_topology); EXPECT_CALL(d_callback, onAsyncWrite(_, _)) .WillOnce(SaveArg<1>(&callbackFn)); - channel->open(); + channel->processReceived(rmqamqp::Message( + rmqamqpt::Method(rmqamqpt::ChannelMethod(rmqamqpt::ChannelClose( + rmqamqpt::Constants::REPLY_SUCCESS, "Closing Channel"))))); } // This shouldn't segfault diff --git a/src/tests/rmqamqp/rmqamqp_receivechannel.t.cpp b/src/tests/rmqamqp/rmqamqp_receivechannel.t.cpp index 984e719f..266080b5 100644 --- a/src/tests/rmqamqp/rmqamqp_receivechannel.t.cpp +++ b/src/tests/rmqamqp/rmqamqp_receivechannel.t.cpp @@ -2264,6 +2264,25 @@ TEST_F(ReceiveChannelTests, ServerCancelAfterChannelCloseSentIsDiscarded) Eq(rmqamqp::Channel::CLEANUP)); } +TEST_F(ReceiveChannelTests, ConsumeOkCrossingGracefulCloseStillCleansUp) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + EXPECT_CALL(d_callback, onAsyncWrite(_, _)).WillOnce(InvokeArgument<1>()); + setupConsumerNoReply(*receiveChannel, d_consumerTag); + + expectChannelClose(); + receiveChannel->gracefulClose(); + + consumerReply(*receiveChannel, d_consumerTag); + + EXPECT_THAT( + receiveChannel->processReceived(rmqamqp::Message(rmqamqpt::Method( + rmqamqpt::ChannelMethod(rmqamqpt::ChannelCloseOk())))), + Eq(rmqamqp::Channel::CLEANUP)); +} + TEST_F(ReceiveChannelTests, ResumeRefusedWhilstClosingForGood) { bsl::shared_ptr receiveChannel = makeReceiveChannel(1); diff --git a/src/tests/rmqamqp/rmqamqp_sendchannel.t.cpp b/src/tests/rmqamqp/rmqamqp_sendchannel.t.cpp index 0076bf68..f98eab04 100644 --- a/src/tests/rmqamqp/rmqamqp_sendchannel.t.cpp +++ b/src/tests/rmqamqp/rmqamqp_sendchannel.t.cpp @@ -515,6 +515,24 @@ TEST_F(SendChannelTests, MessagesArePublishedAfterReconnecting) startupExpectations(*d_sendChannel); } +TEST_F(SendChannelTests, RetryScheduledBeforeResetDoesNotReopen) +{ + startupExpectations(*d_sendChannel); + + rmqio::RetryHandler::RetryCallback scheduledRetry; + EXPECT_CALL(*d_retryHandler, retry(_)) + .WillOnce(SaveArg<0>(&scheduledRetry)); + EXPECT_CALL(d_callback, onAsyncWrite(_, _)).WillOnce(InvokeArgument<1>()); + d_sendChannel->processReceived(rmqamqp::Message( + rmqamqpt::Method(rmqamqpt::ChannelMethod(rmqamqpt::ChannelClose( + rmqamqpt::Constants::REPLY_SUCCESS, "Closing Channel"))))); + + d_sendChannel->reset(); + + EXPECT_CALL(d_callback, onAsyncWrite(_, _)).Times(0); + scheduledRetry(); +} + TEST_F(SendChannelTests, MessagesArePublishedAfterFlow) { rmqt::Message message;