diff --git a/docs/consumer-pause-resume.md b/docs/consumer-pause-resume.md new file mode 100644 index 00000000..33a6a9b0 --- /dev/null +++ b/docs/consumer-pause-resume.md @@ -0,0 +1,71 @@ +# Consumer Pause and Resume + +Consumers created with `consumeOnlyFromHealthyHost` are paused and resumed +by the `HostHealthMonitor`. On every poll that returns a health result +(not a retry) it tells each health-aware consumer to pause (host unhealthy) +or resume (host healthy). + +Pausing sends `basic.cancel` and resuming sends `basic.consume` with the +same consumer tag. The consumer object, and its message callback, are kept +while paused. + +## Pause and resume set the desired state + +`pause()` and `resume()` tell the channel what state the consumer should be +in. The channel makes it so at the next opportunity: straight away if it +can, otherwise once a reply in flight arrives or the channel reopens. The +latest request wins. + +A request is only refused if it repeats the state the consumer is already +in, or if it can't be carried out: + +- Pausing a paused consumer, or resuming a consuming (or starting) one, + resolves with an error and changes nothing. +- Resuming or pausing a cancelled consumer, or one being cancelled, resolves + with an error. Cancel is final. +- Resuming or pausing with no consumer resolves with an error. A consumer + added later starts in the requested state. +- Resuming or pausing a channel that is permanently closing + (`gracefulClose()`) resolves with an error, since it won't reopen and the + broker discards anything sent after `channel.close`. + +Repeating a request that is still in flight returns the same future. + +## When the future resolves + +| Request | Resolves | +|---|---| +| Pause, consumer consuming | On `basic.cancel-ok` | +| Pause, consumer starting | On `basic.cancel-ok` for the `basic.cancel` sent once `basic.consume-ok` arrives | +| Pause, nothing consuming on the broker (e.g. channel reopening) | Immediately; the channel reopens paused | +| Resume, channel ready | On `basic.consume-ok` | +| Resume, channel closed or reopening | On `basic.consume-ok` after the channel reopens | +| Resume, pause in flight | On `basic.consume-ok` for the `basic.consume` sent once the pause's `basic.cancel-ok` arrives | + +A request overtaken by the opposite one before it completes resolves with +an error: + +- A pending resume fails as soon as a pause or cancel is requested. +- A pending pause fails when its `basic.cancel-ok` arrives, if a resume was + requested meanwhile; the resume's `basic.consume` follows straight away. + +Futures still waiting at a channel reset resolve with an error, but the +latest request stands: the reopened channel carries it out. + +## Reopening a channel + +A channel that resets reopens in the state the consumer was last asked to +be in: + +| Latest request at reset | Reopens | +|---|---| +| Resume (or none since consuming) | Consuming | +| Pause | Paused | +| Cancel | Consumer removed | + +## Cancelling + +Cancelling a paused consumer, or one waiting for the channel to reopen, +completes immediately, since nothing is consuming on the broker. Cancelling +while a pause is in flight turns the pause into a cancel; the future +resolves when `basic.cancel-ok` arrives. diff --git a/src/rmq/rmqamqp/rmqamqp_channel.h b/src/rmq/rmqamqp/rmqamqp_channel.h index 8962bb2d..663ba7cd 100644 --- a/src/rmq/rmqamqp/rmqamqp_channel.h +++ b/src/rmq/rmqamqp/rmqamqp_channel.h @@ -164,6 +164,10 @@ class Channel : public bsl::enable_shared_from_this { const bsl::string& vhostName() const; + /// Return true once gracefulClose() has been called: the channel will + /// not reopen + bool isPermanentlyClosing() const { return d_permanentlyClosing; } + virtual const char* channelType() const = 0; virtual bsl::vector > diff --git a/src/rmq/rmqamqp/rmqamqp_receivechannel.cpp b/src/rmq/rmqamqp/rmqamqp_receivechannel.cpp index b61cc05e..c1bbe949 100644 --- a/src/rmq/rmqamqp/rmqamqp_receivechannel.cpp +++ b/src/rmq/rmqamqp/rmqamqp_receivechannel.cpp @@ -100,7 +100,12 @@ class ReceiveChannel::Consumer { const bsl::string& tag, const bool channelPausedOnOpen); + /// True unless NOT_CONSUMING. A channel reset returns a consumer to + /// NOT_CONSUMING, and one created paused starts PAUSED, so this does not + /// mean the consumer has ever consumed. bool isStarted() const; + bool isNotConsuming() const; + bool isStarting() const; bool isActive() const; bool isPaused() const; bool isPausing() const; @@ -183,6 +188,16 @@ bool ReceiveChannel::Consumer::isStarted() const return d_state != NOT_CONSUMING; } +bool ReceiveChannel::Consumer::isNotConsuming() const +{ + return d_state == NOT_CONSUMING; +} + +bool ReceiveChannel::Consumer::isStarting() const +{ + return d_state == STARTING; +} + bool ReceiveChannel::Consumer::isActive() const { return d_state == CONSUMING; } bool ReceiveChannel::Consumer::isPaused() const { return d_state == PAUSED; } @@ -298,6 +313,15 @@ ReceiveChannel::Consumer::cancel(const bool pause) d_state = CANCELLING; } + if (pause && d_state == NOT_CONSUMING) { + // No Basic.Consume has been sent (e.g. the channel is reopening), so + // there is nothing to cancel and no CancelOk will arrive + BALL_LOG_INFO << "Pausing consumer: " << d_tag + << " which is not consuming, queue: " << queueName(); + d_state = PAUSED; + return method; + } + if (d_state != CONSUMING) { BALL_LOG_WARN << (pause ? "Pause" : "Cancel") << " called in state other than CONSUMING. State: " @@ -395,6 +419,8 @@ void ReceiveChannel::onOpen() void ReceiveChannel::onReset() { d_nextMessage.reset(); + d_pauseStartTime = bsls::TimeInterval(); + d_resumeStartTime = bsls::TimeInterval(); d_multipleAckHandler.reset(); // Drop and lock all pending acknowledgement batches @@ -483,7 +509,15 @@ rmqt::Result<> ReceiveChannel::consume(const rmqt::QueueHandle& queue, rmqt::Future<> ReceiveChannel::resume() { - d_channelPausedOnOpen = false; + if (isPermanentlyClosing()) { + return rmqt::Future<>(rmqt::Result<>( + "Resume called on a channel that is permanently closing")); + } + + if (d_consumer && !d_consumer->shouldRestart()) { + return rmqt::Future<>( + rmqt::Result<>("Resume called with a cancel already in flight")); + } if (d_consumeFuturePair) { BALL_LOG_WARN << "Resume called with a resume already in flight"; @@ -491,41 +525,55 @@ rmqt::Future<> ReceiveChannel::resume() } if (!d_consumer) { + // A consumer added later starts consuming + d_channelPausedOnOpen = false; return rmqt::Future<>( rmqt::Result<>("Resume called with no existing consumer")); } - if (d_cancelFuturePair) { - return rmqt::Future<>(rmqt::Result<>( - "Resume called with a " + - bsl::string(d_consumer->isPausing() ? "pause" : "cancel") + - " already in flight")); - } - - if (!d_consumer->isPaused()) { + if (d_consumer->isActive() || d_consumer->isStarting()) { return rmqt::Future<>( rmqt::Result<>("Resume called on unpaused consumer")); } - if (state() == READY || state() == AWAITING_REPLY) { + // From here the resume is accepted, and takes effect at the next + // opportunity. Only count it if it undoes a pause, rather than landing + // whilst a consumer that was never paused reconnects. + const bool wasPaused = d_channelPausedOnOpen || d_consumer->isPaused() || + d_consumer->isPausing(); + d_channelPausedOnOpen = false; + + if (wasPaused) { // Track resume start time and increment counter d_resumeStartTime = bsls::SystemTime::nowMonotonicClock(); d_metricPublisher->publishCounter( Metrics::CONSUMER_RESUME_TOTAL, 1.0, getMetricTags()); + } - d_consumeFuturePair = - bslma::ManagedPtrUtil::makeManaged::Pair>( - rmqt::Future<>::make()); + d_consumeFuturePair = + bslma::ManagedPtrUtil::makeManaged::Pair>( + rmqt::Future<>::make()); + if (d_consumer->isPaused()) { + if (state() == READY || state() == AWAITING_REPLY) { + restartConsumers(); + } + else { + BALL_LOG_TRACE << "Resume called before channel ready, " + "consumertag: " + << d_consumer->consumerTag() + << ", consuming once the channel reopens"; + d_consumer->reset(); + } + } + else if (d_consumer->isNotConsuming() && + (state() == READY || state() == AWAITING_REPLY)) { restartConsumers(); - - return d_consumeFuturePair->second; } + // Otherwise the consumer is NOT_CONSUMING, and is started when the channel + // reopens, or PAUSING, and is restarted when CancelOk arrives - BALL_LOG_TRACE << "Resume called before channel ready, consumertag: " - << d_consumer->consumerTag() << " pending"; - - return rmqt::Future<>(rmqt::Result<>("Resume called before channel ready")); + return d_consumeFuturePair->second; } ReceiveChannel::~ReceiveChannel() @@ -599,6 +647,8 @@ bool ReceiveChannel::consumerIsPaused() const rmqt::Future<> ReceiveChannel::cancel() { + failPendingResume("Consumer cancelled before it could be started"); + if (d_cancelFuturePair) { if (d_consumer->isPausing()) { BALL_LOG_WARN << "Cancel called with a pause already in " @@ -645,14 +695,20 @@ rmqt::Future<> ReceiveChannel::cancel() rmqt::Future<> ReceiveChannel::pause() { + if (isPermanentlyClosing()) { + return rmqt::Future<>(rmqt::Result<>( + "Pause called on a channel that is permanently closing")); + } + + if (d_consumer && !d_consumer->shouldRestart()) { + return rmqt::Future<>( + rmqt::Result<>("Pause called with a cancel already in flight")); + } + d_channelPausedOnOpen = true; + failPendingResume("Resume superseded by pause"); if (d_cancelFuturePair) { - if (d_consumer->isCancelling()) { - return rmqt::Future<>( - rmqt::Result<>("Pause called with a cancel already in flight")); - } - BALL_LOG_WARN << "Pause called with a pause already in flight"; return d_cancelFuturePair->second; } @@ -663,18 +719,25 @@ rmqt::Future<> ReceiveChannel::pause() } if (d_consumer->canPause()) { - // Track pause start time and increment counter - d_pauseStartTime = bsls::SystemTime::nowMonotonicClock(); d_metricPublisher->publishCounter( Metrics::CONSUMER_PAUSE_TOTAL, 1.0, getMetricTags()); + const bool pause = true; + bsl::optional method = d_consumer->cancel(pause); + + if (d_consumer->isPaused()) { + // Nothing was consuming on the broker, so the pause has already + // taken effect + return rmqt::Future<>(rmqt::Result<>()); + } + + // Track pause start time + d_pauseStartTime = bsls::SystemTime::nowMonotonicClock(); + d_cancelFuturePair = bslma::ManagedPtrUtil::makeManaged::Pair>( rmqt::Future<>::make()); - const bool pause = true; - bsl::optional method = d_consumer->cancel(pause); - if (method) { writeMessage(Message(rmqamqpt::Method(method.value())), AWAITING_REPLY); @@ -682,8 +745,7 @@ rmqt::Future<> ReceiveChannel::pause() return d_cancelFuturePair->second; } - BALL_LOG_DEBUG << "Pause called when consumer is already paused"; - return rmqt::Future<>(rmqt::Result<>()); + return rmqt::Future<>(rmqt::Result<>("Pause called on paused consumer")); } rmqt::Future<> ReceiveChannel::drain() @@ -754,7 +816,9 @@ void ReceiveChannel::processBasicMethod(const rmqamqpt::BasicMethod& basic) AWAITING_REPLY); } - if (d_consumeFuturePair) { + // A resume requested after a pause waits for the pause's + // CancelOk and the Basic.Consume that follows it + if (d_consumeFuturePair && d_consumer->isActive()) { // Publish resume latency publishLatencyIfTracked( &d_resumeStartTime, @@ -779,7 +843,7 @@ void ReceiveChannel::processBasicMethod(const rmqamqpt::BasicMethod& basic) } break; case rmqamqpt::BasicQoSOk::METHOD_ID: { if (d_consumer) { - if (d_channelPausedOnOpen) { + if (d_channelPausedOnOpen && d_consumer->isNotConsuming()) { d_consumer->setPaused(); } @@ -789,11 +853,14 @@ void ReceiveChannel::processBasicMethod(const rmqamqpt::BasicMethod& basic) d_hungProgressTimer->cancel(); ready(); } - else { + else if (d_consumer->isNotConsuming()) { // we've restarted, we already have a callback so can // declare consumer restartConsumers(); } + // Otherwise a resume issued whilst the QoS was in flight has + // already sent Basic.Consume, and its reply will mark the + // channel ready } else { // This is the first connection, or recovery from connection @@ -822,6 +889,10 @@ void ReceiveChannel::processBasicMethod(const rmqamqpt::BasicMethod& basic) basic.the().consumerTag()) { d_consumer->cancelOk(); + // A resume requested whilst the pause was in flight + const bool resumeRequested = + d_consumer->isPaused() && !d_channelPausedOnOpen; + if (d_consumer->isPaused()) { BALL_LOG_INFO << "Pausing Consumer: " << d_consumer->consumerTag(); @@ -834,12 +905,18 @@ void ReceiveChannel::processBasicMethod(const rmqamqpt::BasicMethod& basic) d_consumer.reset(); } - if (d_consumeFuturePair) { + if (d_consumeFuturePair && !resumeRequested) { d_consumeFuturePair->first( rmqt::Result<>("Consumer cancelled before it could be " "started")); d_consumeFuturePair.reset(); } + if (d_cancelFuturePair && resumeRequested) { + d_pauseStartTime = bsls::TimeInterval(); + d_cancelFuturePair->first( + rmqt::Result<>("Pause superseded by resume")); + d_cancelFuturePair.reset(); + } if (d_cancelFuturePair) { // Publish pause latency publishLatencyIfTracked( @@ -855,6 +932,10 @@ void ReceiveChannel::processBasicMethod(const rmqamqpt::BasicMethod& basic) // Ensure any waitForReady() futures are resolved before we // enter the fully cancelled state ready(); + + if (resumeRequested) { + restartConsumers(); + } } else { invalidConsumerError( @@ -894,6 +975,15 @@ void ReceiveChannel::processMessage(const rmqt::Message& message) } } +void ReceiveChannel::failPendingResume(const bsl::string& reason) +{ + if (d_consumeFuturePair) { + d_resumeStartTime = bsls::TimeInterval(); + d_consumeFuturePair->first(rmqt::Result<>(reason)); + d_consumeFuturePair.reset(); + } +} + void ReceiveChannel::restartConsumers() { if (d_consumer) { diff --git a/src/rmq/rmqamqp/rmqamqp_receivechannel.h b/src/rmq/rmqamqp/rmqamqp_receivechannel.h index 6ec2bce4..754e6be0 100644 --- a/src/rmq/rmqamqp/rmqamqp_receivechannel.h +++ b/src/rmq/rmqamqp/rmqamqp_receivechannel.h @@ -92,7 +92,9 @@ class ReceiveChannel : public Channel { const MessageCallback& onNewMessage, const bsl::string& consumerTag = ""); - /// Resumes the paused consumer on the channel + /// Resumes the paused consumer on the channel at the next opportunity: + /// immediately if the channel is ready, otherwise once it reopens or a + /// pause in flight completes. /// returns Future that will resolve when ConsumeOk received from server virtual rmqt::Future<> resume(); @@ -111,7 +113,9 @@ class ReceiveChannel : public Channel { virtual rmqt::Future<> cancel(); /// Pauses the active consumer on the channel - /// returns Future that will resolve when CancelOk received from server + /// returns Future that will resolve when CancelOk received from server, + /// or immediately if the consumer is not consuming on the server (e.g. + /// whilst the channel is reopening) virtual rmqt::Future<> pause(); /// If the channel is in a cancelled state, waits for number of the @@ -147,6 +151,7 @@ class ReceiveChannel : public Channel { void onReset() BSLS_KEYWORD_OVERRIDE; void processFailures() BSLS_KEYWORD_OVERRIDE; void restartConsumers(); + void failPendingResume(const bsl::string& reason); void invalidConsumerError(const bsl::string& consumerTag); void removeSingleMessageFromStore(uint64_t deliveryTag); void removeMultipleMessagesFromStore(uint64_t deliveryTag); diff --git a/src/tests/rmqamqp/rmqamqp_receivechannel.t.cpp b/src/tests/rmqamqp/rmqamqp_receivechannel.t.cpp index e7ca8c66..5b602312 100644 --- a/src/tests/rmqamqp/rmqamqp_receivechannel.t.cpp +++ b/src/tests/rmqamqp/rmqamqp_receivechannel.t.cpp @@ -289,6 +289,25 @@ class ReceiveChannelTests : public rmqamqp::ChannelTests { receiveBasicQoSOk(rc); } + /// Reset the channel and reopen it as far as sending Basic.QoS + void resetAndReopenUntilQoSInFlight(rmqamqp::ReceiveChannel& rc) + { + rc.reset(); + openAndSendTopology(rc); + expectBasicQoS(); + queueDeclareReply(rc); + EXPECT_THAT(rc.state(), Eq(rmqamqp::Channel::AWAITING_REPLY)); + } + + void setupPausedConsumer(rmqamqp::ReceiveChannel& rc) + { + setupConsumer(rc, d_consumerTag); + expectBasicCancel(); + rc.pause(); + receiveBasicCancelOk(rc, d_consumerTag); + EXPECT_TRUE(rc.consumerIsPaused()); + } + void receiveMessage(rmqamqp::ReceiveChannel& rc, uint64_t deliveryTag, const bsl::string& consumerTag = "") @@ -1001,14 +1020,15 @@ TEST_F(ReceiveChannelTests, ClientCancelDuringResume) EXPECT_FALSE(receiveChannel->consumerIsActive()); EXPECT_FALSE(receiveChannel->consumerIsPaused()); - // Cancel before resume reply + // Cancel before resume reply supersedes the resume notExpectBasicCancel(); rmqt::Future<> wereCancelled = receiveChannel->cancel(); + EXPECT_FALSE(wereResumed.tryResult()); + EXPECT_THAT(wereResumed.tryResult().returnCode(), Ne(rmqt::TIMEOUT)); // Complete resume and expect cancel to be sent afterwards expectBasicCancel(); receiveBasicConsumeOk(*receiveChannel, d_consumerTag); - EXPECT_TRUE(!!wereResumed.tryResult()); receiveBasicCancelOk(*receiveChannel, d_consumerTag); EXPECT_TRUE(!!wereCancelled.tryResult()); @@ -1501,15 +1521,16 @@ TEST_F(ReceiveChannelTests, PauseDuringResume) expectBasicConsume(); rmqt::Future<> wereResumed = receiveChannel->resume(); - // Start pausing + // Start pausing: supersedes the resume expectBasicCancel(); werePaused = receiveChannel->pause(); EXPECT_FALSE(!!werePaused.tryResult()); EXPECT_FALSE(receiveChannel->consumerIsPaused()); + EXPECT_THAT(wereResumed.tryResult().error(), + Eq("Resume superseded by pause")); // Complete resume receiveBasicConsumeOk(*receiveChannel, d_consumerTag); - EXPECT_TRUE(!!wereResumed.tryResult()); EXPECT_FALSE(receiveChannel->consumerIsActive()); EXPECT_FALSE(receiveChannel->consumerIsPaused()); @@ -1598,10 +1619,11 @@ TEST_F(ReceiveChannelTests, PauseAfterPause) EXPECT_FALSE(receiveChannel->consumerIsActive()); EXPECT_TRUE(receiveChannel->consumerIsPaused()); - // Pause again + // Pause again: already paused, so the duplicate request is refused notExpectBasicCancel(); rmqt::Future<> werePaused2x = receiveChannel->pause(); - EXPECT_TRUE(!!werePaused2x.tryResult()); + EXPECT_THAT(werePaused2x.tryResult().error(), + Eq("Pause called on paused consumer")); EXPECT_FALSE(receiveChannel->consumerIsActive()); EXPECT_TRUE(receiveChannel->consumerIsPaused()); } @@ -1690,6 +1712,511 @@ TEST_F(ReceiveChannelTests, PausePersistsOnReset) EXPECT_TRUE(receiveChannel->consumerIsPaused()); } +TEST_F(ReceiveChannelTests, PauseDuringReopenCompletesImmediately) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupConsumer(*receiveChannel, d_consumerTag); + EXPECT_TRUE(receiveChannel->consumerIsActive()); + + // Disconnect: nothing is consuming on the broker until the channel reopens + receiveChannel->reset(); + + // Pause while reopening: there is no consume to cancel, so no CancelOk will + // arrive and the pause must complete straight away + notExpectBasicCancel(); + rmqt::Future<> werePaused = receiveChannel->pause(); + EXPECT_TRUE(werePaused.tryResult()); + EXPECT_TRUE(receiveChannel->consumerIsPaused()); + + // Reconnect paused + makeReady(*receiveChannel); + EXPECT_THAT(receiveChannel->state(), Eq(rmqamqp::Channel::READY)); + EXPECT_TRUE(receiveChannel->consumerIsPaused()); + + // Resume is not blocked by a pause that will never complete + expectBasicConsume(); + rmqt::Future<> wereResumed = receiveChannel->resume(); + receiveBasicConsumeOk(*receiveChannel, d_consumerTag); + EXPECT_TRUE(wereResumed.tryResult()); + EXPECT_TRUE(receiveChannel->consumerIsActive()); +} + +TEST_F(ReceiveChannelTests, PauseThenResumeDuringReopenDoesNotHang) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupConsumer(*receiveChannel, d_consumerTag); + EXPECT_TRUE(receiveChannel->consumerIsActive()); + + receiveChannel->reset(); + + // Host goes unhealthy, then healthy again, before the channel reopens + receiveChannel->pause(); + rmqt::Future<> wereResumed = receiveChannel->resume(); + EXPECT_THAT(wereResumed.tryResult().returnCode(), Eq(rmqt::TIMEOUT)); + + // The latest request wins: the reopened channel consumes, rather than + // waiting for the hung timer to tear down the connection + expectBasicConsume(); + makeReady(*receiveChannel); + receiveBasicConsumeOk(*receiveChannel, d_consumerTag); + EXPECT_THAT(receiveChannel->state(), Eq(rmqamqp::Channel::READY)); + EXPECT_TRUE(wereResumed.tryResult()); + EXPECT_TRUE(receiveChannel->consumerIsActive()); +} + +TEST_F(ReceiveChannelTests, CancelAfterPauseDuringReopen) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupConsumer(*receiveChannel, d_consumerTag); + + receiveChannel->reset(); + receiveChannel->pause(); + makeReady(*receiveChannel); + EXPECT_TRUE(receiveChannel->consumerIsPaused()); + + // Nothing is consuming on the broker, so cancel completes straight away + notExpectBasicCancel(); + rmqt::Future<> wereCancelled = receiveChannel->cancel(); + EXPECT_TRUE(wereCancelled.tryResult()); + EXPECT_FALSE(receiveChannel->consumerIsActive()); + EXPECT_FALSE(receiveChannel->consumerIsPaused()); +} + +TEST_F(ReceiveChannelTests, CancelBeforeReopenAfterPauseDuringReopen) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupConsumer(*receiveChannel, d_consumerTag); + + receiveChannel->reset(); + receiveChannel->pause(); + + notExpectBasicCancel(); + rmqt::Future<> wereCancelled = receiveChannel->cancel(); + EXPECT_TRUE(wereCancelled.tryResult()); + + // Reopening must not bring the cancelled consumer back as paused + makeReady(*receiveChannel); + EXPECT_THAT(receiveChannel->state(), Eq(rmqamqp::Channel::READY)); + EXPECT_FALSE(receiveChannel->consumerIsActive()); + EXPECT_FALSE(receiveChannel->consumerIsPaused()); + + rmqt::Future<> wereResumed = receiveChannel->resume(); + EXPECT_THAT(wereResumed.tryResult().error(), + Eq("Resume called with no existing consumer")); +} + +TEST_F(ReceiveChannelTests, ResumeThenPauseWhileQoSInFlight) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupPausedConsumer(*receiveChannel); + resetAndReopenUntilQoSInFlight(*receiveChannel); + + // Resume is sent ahead of QoSOk, then a pause follows it + expectBasicConsume(); + receiveChannel->resume(); + rmqt::Future<> werePaused = receiveChannel->pause(); + EXPECT_THAT(werePaused.tryResult().returnCode(), Eq(rmqt::TIMEOUT)); + + // QoSOk must not mark the consumer paused while a consume is on the wire + receiveBasicQoSOk(*receiveChannel); + EXPECT_FALSE(receiveChannel->consumerIsPaused()); + + // ConsumeOk turns into a cancel, and CancelOk completes the pause + expectBasicCancel(); + receiveBasicConsumeOk(*receiveChannel, d_consumerTag); + EXPECT_FALSE(receiveChannel->consumerIsPaused()); + + receiveBasicCancelOk(*receiveChannel, d_consumerTag); + EXPECT_TRUE(werePaused.tryResult()); + EXPECT_TRUE(receiveChannel->consumerIsPaused()); + EXPECT_THAT(receiveChannel->state(), Eq(rmqamqp::Channel::READY)); +} + +TEST_F(ReceiveChannelTests, ResumeWhileQoSInFlight) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupPausedConsumer(*receiveChannel); + resetAndReopenUntilQoSInFlight(*receiveChannel); + + // The resume's Basic.Consume, sent ahead of QoSOk, starts the consumer + expectBasicConsume(); + rmqt::Future<> wereResumed = receiveChannel->resume(); + receiveBasicQoSOk(*receiveChannel); + EXPECT_THAT(wereResumed.tryResult().returnCode(), Eq(rmqt::TIMEOUT)); + + receiveBasicConsumeOk(*receiveChannel, d_consumerTag); + EXPECT_TRUE(wereResumed.tryResult()); + EXPECT_TRUE(receiveChannel->consumerIsActive()); + EXPECT_THAT(receiveChannel->state(), Eq(rmqamqp::Channel::READY)); +} + +TEST_F(ReceiveChannelTests, ResumeThenCancelWhileQoSInFlight) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupPausedConsumer(*receiveChannel); + resetAndReopenUntilQoSInFlight(*receiveChannel); + + expectBasicConsume(); + receiveChannel->resume(); + rmqt::Future<> wereCancelled = receiveChannel->cancel(); + receiveBasicQoSOk(*receiveChannel); + + // ConsumeOk turns into a cancel, and CancelOk removes the consumer + expectBasicCancel(); + receiveBasicConsumeOk(*receiveChannel, d_consumerTag); + receiveBasicCancelOk(*receiveChannel, d_consumerTag); + EXPECT_TRUE(wereCancelled.tryResult()); + EXPECT_FALSE(receiveChannel->consumerIsActive()); + EXPECT_FALSE(receiveChannel->consumerIsPaused()); + EXPECT_THAT(receiveChannel->state(), Eq(rmqamqp::Channel::READY)); +} + +TEST_F(ReceiveChannelTests, PauseWhileQoSInFlightCompletesImmediately) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupConsumer(*receiveChannel, d_consumerTag); + resetAndReopenUntilQoSInFlight(*receiveChannel); + + notExpectBasicCancel(); + rmqt::Future<> werePaused = receiveChannel->pause(); + EXPECT_TRUE(werePaused.tryResult()); + EXPECT_TRUE(receiveChannel->consumerIsPaused()); + + receiveBasicQoSOk(*receiveChannel); + EXPECT_THAT(receiveChannel->state(), Eq(rmqamqp::Channel::READY)); + EXPECT_TRUE(receiveChannel->consumerIsPaused()); +} + +TEST_F(ReceiveChannelTests, HungConsumeOkAfterResumeWhileQoSInFlight) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupPausedConsumer(*receiveChannel); + resetAndReopenUntilQoSInFlight(*receiveChannel); + + expectBasicConsume(); + receiveChannel->resume(); + receiveBasicQoSOk(*receiveChannel); + + // If the channel is never sent a ConsumeOk, the hung callback should be + // called. + const int HUNG_TIMEOUT_SECONDS = 65; + EXPECT_CALL(d_callback, onHungTimerCallback()); + d_timerFactory->step_time(bsls::TimeInterval(HUNG_TIMEOUT_SECONDS)); +} + +TEST_F(ReceiveChannelTests, ResumeDuringReopen) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupPausedConsumer(*receiveChannel); + receiveChannel->reset(); + + // Resume whilst the channel is closed takes effect once it reopens + rmqt::Future<> wereResumed = receiveChannel->resume(); + EXPECT_THAT(wereResumed.tryResult().returnCode(), Eq(rmqt::TIMEOUT)); + + expectBasicConsume(); + makeReady(*receiveChannel); + EXPECT_THAT(wereResumed.tryResult().returnCode(), Eq(rmqt::TIMEOUT)); + + receiveBasicConsumeOk(*receiveChannel, d_consumerTag); + EXPECT_TRUE(wereResumed.tryResult()); + EXPECT_TRUE(receiveChannel->consumerIsActive()); + EXPECT_THAT(receiveChannel->state(), Eq(rmqamqp::Channel::READY)); +} + +TEST_F(ReceiveChannelTests, ResumeDuringReopenAfterPauseInFlightAtReset) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupConsumer(*receiveChannel, d_consumerTag); + + // Pause sent, channel resets before CancelOk + expectBasicCancel(); + receiveChannel->pause(); + receiveChannel->reset(); + + rmqt::Future<> wereResumed = receiveChannel->resume(); + EXPECT_THAT(wereResumed.tryResult().returnCode(), Eq(rmqt::TIMEOUT)); + + expectBasicConsume(); + makeReady(*receiveChannel); + receiveBasicConsumeOk(*receiveChannel, d_consumerTag); + EXPECT_TRUE(wereResumed.tryResult()); + EXPECT_TRUE(receiveChannel->consumerIsActive()); +} + +TEST_F(ReceiveChannelTests, ResumeThenPauseDuringReopen) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupPausedConsumer(*receiveChannel); + receiveChannel->reset(); + + rmqt::Future<> wereResumed = receiveChannel->resume(); + rmqt::Future<> werePaused = receiveChannel->pause(); + + // The pause supersedes the resume before it took effect + EXPECT_TRUE(werePaused.tryResult()); + EXPECT_FALSE(wereResumed.tryResult()); + EXPECT_THAT(wereResumed.tryResult().returnCode(), Ne(rmqt::TIMEOUT)); + + notExpectBasicConsume(); + makeReady(*receiveChannel); + EXPECT_THAT(receiveChannel->state(), Eq(rmqamqp::Channel::READY)); + EXPECT_TRUE(receiveChannel->consumerIsPaused()); +} + +TEST_F(ReceiveChannelTests, CancelAfterResumeDuringReopen) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupPausedConsumer(*receiveChannel); + receiveChannel->reset(); + + rmqt::Future<> wereResumed = receiveChannel->resume(); + rmqt::Future<> wereCancelled = receiveChannel->cancel(); + EXPECT_TRUE(wereCancelled.tryResult()); + EXPECT_FALSE(wereResumed.tryResult()); + EXPECT_THAT(wereResumed.tryResult().returnCode(), Ne(rmqt::TIMEOUT)); + + notExpectBasicConsume(); + makeReady(*receiveChannel); + EXPECT_THAT(receiveChannel->state(), Eq(rmqamqp::Channel::READY)); + EXPECT_FALSE(receiveChannel->consumerIsActive()); + EXPECT_FALSE(receiveChannel->consumerIsPaused()); +} + +TEST_F(ReceiveChannelTests, ResumePauseResumeWhileConsumeInFlight) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupPausedConsumer(*receiveChannel); + + expectBasicConsume(); + rmqt::Future<> wereResumed = receiveChannel->resume(); + rmqt::Future<> werePaused = receiveChannel->pause(); + EXPECT_THAT(wereResumed.tryResult().error(), + Eq("Resume superseded by pause")); + + // The latest request is resume + rmqt::Future<> wereResumed2 = receiveChannel->resume(); + EXPECT_THAT(wereResumed2.tryResult().returnCode(), Eq(rmqt::TIMEOUT)); + + // ConsumeOk for the first resume: the pause still goes out, and the + // second resume waits for it + expectBasicCancel(); + receiveBasicConsumeOk(*receiveChannel, d_consumerTag); + EXPECT_THAT(wereResumed2.tryResult().returnCode(), Eq(rmqt::TIMEOUT)); + + expectBasicConsume(); + receiveBasicCancelOk(*receiveChannel, d_consumerTag); + EXPECT_THAT(werePaused.tryResult().error(), + Eq("Pause superseded by resume")); + EXPECT_THAT(wereResumed2.tryResult().returnCode(), Eq(rmqt::TIMEOUT)); + + receiveBasicConsumeOk(*receiveChannel, d_consumerTag); + EXPECT_TRUE(wereResumed2.tryResult()); + EXPECT_TRUE(receiveChannel->consumerIsActive()); + EXPECT_THAT(receiveChannel->state(), Eq(rmqamqp::Channel::READY)); +} + +TEST_F(ReceiveChannelTests, ResumeAfterPauseBeforeFirstConsumeOk) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + + expectBasicConsume(); + setupConsumerNoReply(*receiveChannel); + rmqt::Future<> werePaused = receiveChannel->pause(); + rmqt::Future<> wereResumed = receiveChannel->resume(); + + // The ConsumeOk belongs to the consume sent before the pause, so it does + // not complete the resume + expectBasicCancel(); + receiveBasicConsumeOk(*receiveChannel, d_consumerTag); + EXPECT_THAT(wereResumed.tryResult().returnCode(), Eq(rmqt::TIMEOUT)); + + expectBasicConsume(); + receiveBasicCancelOk(*receiveChannel, d_consumerTag); + EXPECT_THAT(werePaused.tryResult().error(), + Eq("Pause superseded by resume")); + + receiveBasicConsumeOk(*receiveChannel, d_consumerTag); + EXPECT_TRUE(wereResumed.tryResult()); + EXPECT_TRUE(receiveChannel->consumerIsActive()); +} + +TEST_F(ReceiveChannelTests, RefusedPauseDoesNotPauseNextConsumer) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupConsumer(*receiveChannel, d_consumerTag); + + expectBasicCancel(); + rmqt::Future<> wereCancelled = receiveChannel->cancel(); + rmqt::Future<> werePaused = receiveChannel->pause(); + EXPECT_THAT(werePaused.tryResult().error(), + Eq("Pause called with a cancel already in flight")); + + receiveBasicCancelOk(*receiveChannel, d_consumerTag); + EXPECT_TRUE(wereCancelled.tryResult()); + + // A new consumer is not affected by the refused pause + expectBasicConsume(); + setupConsumerNoReply(*receiveChannel); + receiveBasicConsumeOk(*receiveChannel, d_consumerTag); + EXPECT_TRUE(receiveChannel->consumerIsActive()); +} + +TEST_F(ReceiveChannelTests, ResumeDuringReconnectIsNotCountedAsResume) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupConsumer(*receiveChannel, d_consumerTag); + receiveChannel->reset(); + + EXPECT_CALL(*d_metricPublisher, publishCounter(_, _, _)).Times(AnyNumber()); + EXPECT_CALL(*d_metricPublisher, publishSummary(_, _, _)).Times(AnyNumber()); + + // The consumer was never paused, so a health poll landing mid-reconnect + // is not a resume + EXPECT_CALL(*d_metricPublisher, + publishCounter( + bsl::string(rmqamqp::Metrics::CONSUMER_RESUME_TOTAL), _, _)) + .Times(0); + EXPECT_CALL( + *d_metricPublisher, + publishSummary( + bsl::string(rmqamqp::Metrics::RESUME_OPERATION_LATENCY_SECONDS), + _, + _)) + .Times(0); + + rmqt::Future<> wereResumed = receiveChannel->resume(); + + expectBasicConsume(); + makeReady(*receiveChannel); + receiveBasicConsumeOk(*receiveChannel, d_consumerTag); + EXPECT_TRUE(wereResumed.tryResult()); + EXPECT_TRUE(receiveChannel->consumerIsActive()); +} + +TEST_F(ReceiveChannelTests, SupersededPauseDoesNotReportLatencyOnLaterCancel) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupConsumer(*receiveChannel, d_consumerTag); + + expectBasicCancel(); + receiveChannel->pause(); + receiveChannel->resume(); + expectBasicConsume(); + receiveBasicCancelOk(*receiveChannel, d_consumerTag); + receiveBasicConsumeOk(*receiveChannel, d_consumerTag); + EXPECT_TRUE(receiveChannel->consumerIsActive()); + + EXPECT_CALL(*d_metricPublisher, publishCounter(_, _, _)).Times(AnyNumber()); + EXPECT_CALL(*d_metricPublisher, publishSummary(_, _, _)).Times(AnyNumber()); + EXPECT_CALL( + *d_metricPublisher, + publishSummary( + bsl::string(rmqamqp::Metrics::PAUSE_OPERATION_LATENCY_SECONDS), + _, + _)) + .Times(0); + + expectBasicCancel(); + rmqt::Future<> wereCancelled = receiveChannel->cancel(); + receiveBasicCancelOk(*receiveChannel, d_consumerTag); + EXPECT_TRUE(wereCancelled.tryResult()); +} + +TEST_F(ReceiveChannelTests, CancelAfterResumeDuringPause) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupConsumer(*receiveChannel, d_consumerTag); + + expectBasicCancel(); + receiveChannel->pause(); + rmqt::Future<> wereResumed = receiveChannel->resume(); + + // Cancel supersedes the resume, and turns the pause into a cancel + notExpectBasicConsume(); + rmqt::Future<> wereCancelled = receiveChannel->cancel(); + EXPECT_THAT(wereResumed.tryResult().error(), + Eq("Consumer cancelled before it could be started")); + + receiveBasicCancelOk(*receiveChannel, d_consumerTag); + EXPECT_TRUE(wereCancelled.tryResult()); + EXPECT_FALSE(receiveChannel->consumerIsActive()); + EXPECT_FALSE(receiveChannel->consumerIsPaused()); + EXPECT_THAT(receiveChannel->state(), Eq(rmqamqp::Channel::READY)); +} + +TEST_F(ReceiveChannelTests, PauseRefusedWhilstClosingForGood) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupConsumer(*receiveChannel, d_consumerTag); + + expectChannelClose(); + receiveChannel->gracefulClose(); + + // The broker discards a Basic.Cancel sent after Channel.Close + notExpectBasicCancel(); + rmqt::Future<> werePaused = receiveChannel->pause(); + EXPECT_THAT(werePaused.tryResult().error(), + Eq("Pause called on a channel that is permanently closing")); +} + +TEST_F(ReceiveChannelTests, ResumeRefusedWhilstClosingForGood) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupPausedConsumer(*receiveChannel); + + expectChannelClose(); + receiveChannel->gracefulClose(); + + // The channel won't reopen, so the resume can't be carried out + notExpectBasicConsume(); + rmqt::Future<> wereResumed = receiveChannel->resume(); + EXPECT_THAT(wereResumed.tryResult().error(), + Eq("Resume called on a channel that is permanently closing")); +} + TEST_F(ReceiveChannelTests, PauseFail) { bsl::shared_ptr receiveChannel = makeReceiveChannel(1); @@ -1737,10 +2264,9 @@ TEST_F(ReceiveChannelTests, ResumeDuringConsume) EXPECT_THAT(receiveChannel->state(), Eq(rmqamqp::Channel::READY)); EXPECT_TRUE(receiveChannel->consumerIsActive()); - // Resume + // Resume: already consuming, so the duplicate request is refused notExpectBasicConsume(); rmqt::Future<> wereResumed2x = receiveChannel->resume(); - EXPECT_FALSE(wereResumed2x.tryResult()); EXPECT_THAT(wereResumed2x.tryResult().error(), Eq("Resume called on unpaused consumer")); EXPECT_TRUE(receiveChannel->consumerIsActive()); @@ -1824,10 +2350,9 @@ TEST_F(ReceiveChannelTests, ResumeBeforePause) EXPECT_THAT(receiveChannel->state(), Eq(rmqamqp::Channel::READY)); EXPECT_TRUE(receiveChannel->consumerIsActive()); - // Resume + // Resume: already consuming, so the duplicate request is refused notExpectBasicConsume(); rmqt::Future<> wereResumed = receiveChannel->resume(); - EXPECT_FALSE(!!wereResumed.tryResult()); EXPECT_THAT(wereResumed.tryResult().error(), Eq("Resume called on unpaused consumer")); EXPECT_TRUE(receiveChannel->consumerIsActive()); @@ -1849,12 +2374,20 @@ TEST_F(ReceiveChannelTests, ResumeDuringPause) EXPECT_FALSE(receiveChannel->consumerIsActive()); EXPECT_FALSE(receiveChannel->consumerIsPaused()); - // Start resuming before pause reply - notExpectBasicConsume(); + // Resume before pause reply: takes effect once the pause completes rmqt::Future<> wereResumed = receiveChannel->resume(); - EXPECT_FALSE(!!wereResumed.tryResult()); - EXPECT_THAT(wereResumed.tryResult().error(), - Eq("Resume called with a pause already in flight")); + EXPECT_THAT(wereResumed.tryResult().returnCode(), Eq(rmqt::TIMEOUT)); + + expectBasicConsume(); + receiveBasicCancelOk(*receiveChannel, d_consumerTag); + EXPECT_THAT(werePaused.tryResult().error(), + Eq("Pause superseded by resume")); + EXPECT_THAT(wereResumed.tryResult().returnCode(), Eq(rmqt::TIMEOUT)); + + receiveBasicConsumeOk(*receiveChannel, d_consumerTag); + EXPECT_TRUE(wereResumed.tryResult()); + EXPECT_TRUE(receiveChannel->consumerIsActive()); + EXPECT_THAT(receiveChannel->state(), Eq(rmqamqp::Channel::READY)); } TEST_F(ReceiveChannelTests, ResumeDuringCancel)