diff --git a/src/rmq/rmqamqp/rmqamqp_channel.cpp b/src/rmq/rmqamqp/rmqamqp_channel.cpp index 1307341f..76d18cd5 100644 --- a/src/rmq/rmqamqp/rmqamqp_channel.cpp +++ b/src/rmq/rmqamqp/rmqamqp_channel.cpp @@ -385,6 +385,16 @@ Channel::processReceived(const rmqamqp::Message& message) return KEEP; } + if (d_state == CHANNEL_CLOSE_SENT && + !MessageUtil::isMethod( + message) && + !MessageUtil::isMethod(message)) { + BALL_LOG_DEBUG << "Discarding frame received after sending " + "Channel.Close"; + return KEEP; + } + if (message.is()) { const rmqamqpt::Method& method = message.the(); switch (method.classId()) { diff --git a/src/rmq/rmqamqp/rmqamqp_message.h b/src/rmq/rmqamqp/rmqamqp_message.h index 1fde173c..a9e80399 100644 --- a/src/rmq/rmqamqp/rmqamqp_message.h +++ b/src/rmq/rmqamqp/rmqamqp_message.h @@ -38,6 +38,23 @@ namespace rmqamqp { typedef bdlb::Variant Message; +struct MessageUtil { + template + static bool isMethod(const Message& message); +}; + +template +bool MessageUtil::isMethod(const Message& message) +{ + if (!message.is()) { + return false; + } + + const rmqamqpt::Method& method = message.the(); + return method.is() && + method.the().template is(); +} + } // namespace rmqamqp } // namespace BloombergLP diff --git a/src/tests/rmqamqp/rmqamqp_connection.t.cpp b/src/tests/rmqamqp/rmqamqp_connection.t.cpp index 1484bbd8..d324f424 100644 --- a/src/tests/rmqamqp/rmqamqp_connection.t.cpp +++ b/src/tests/rmqamqp/rmqamqp_connection.t.cpp @@ -125,7 +125,7 @@ class MockConnection : public rmqio::Connection { { } - rmqamqpt::Method examineFrameMethod(const SerializedFrame& frame) + rmqamqp::Message decodeMessage(const SerializedFrame& frame) { rmqamqpt::Frame decoded; size_t unused1 = 0, unused2 = 0; @@ -135,17 +135,16 @@ class MockConnection : public rmqio::Connection { frame.serialized(), frame.frameLength()); if (rc != Frame::OK) { - return rmqamqpt::Method(); + return rmqamqp::Message(); } rmqamqp::Framer framer; uint16_t channel; rmqamqp::Message msg; - if (framer.appendFrame(&channel, &msg, decoded) == - rmqamqp::Framer::OK && - msg.is()) { - return msg.the(); + if (framer.appendFrame(&channel, &msg, decoded) != + rmqamqp::Framer::OK) { + return rmqamqp::Message(); } - return rmqamqpt::Method(); + return msg; } void feedNextFrame() @@ -188,10 +187,10 @@ class MockConnection : public rmqio::Connection { ++it) { BSLS_ASSERT_OPT(*(d_replayFrame.writeFrame()) == **it); } - bool closeOk = rmqamqpt::Method::Util::typeMatch( - examineFrameMethod(*frames[0]), - rmqamqpt::Method( - rmqamqpt::ConnectionMethod(rmqamqpt::ConnectionCloseOk()))); + bool closeOk = + rmqamqp::MessageUtil::isMethod( + decodeMessage(*frames[0])); boost::asio::post(d_eventLoop, callback); diff --git a/src/tests/rmqamqp/rmqamqp_receivechannel.t.cpp b/src/tests/rmqamqp/rmqamqp_receivechannel.t.cpp index 268256ed..984e719f 100644 --- a/src/tests/rmqamqp/rmqamqp_receivechannel.t.cpp +++ b/src/tests/rmqamqp/rmqamqp_receivechannel.t.cpp @@ -69,15 +69,6 @@ const MethodTypeT& MethodGetter(const bsl::shared_ptr& msg) throw bsl::runtime_error("MethodMatcher: No Match"); } -template -bool MethodMatcher(const bsl::shared_ptr& msg) -{ - return (msg->is() && - msg->the().is() && - msg->the() - .the() - .template is()); -} MATCHER_P(EXPECT_QOSPREFETCH_IS, pf, "Prefetch isn't " + PrintToString(pf)) { try { @@ -2254,6 +2245,25 @@ TEST_F(ReceiveChannelTests, PauseRefusedWhilstClosingForGood) Eq("Pause called on a channel that is permanently closing")); } +TEST_F(ReceiveChannelTests, ServerCancelAfterChannelCloseSentIsDiscarded) +{ + bsl::shared_ptr receiveChannel = makeReceiveChannel(1); + + makeReady(*receiveChannel); + setupConsumer(*receiveChannel, d_consumerTag); + + expectChannelClose(); + receiveChannel->gracefulClose(); + + EXPECT_CALL(d_callback, onAsyncWrite(_, _)).Times(0); + receiveBasicCancel(*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_topologytransformer.t.cpp b/src/tests/rmqamqp/rmqamqp_topologytransformer.t.cpp index 41369f31..4c694eec 100644 --- a/src/tests/rmqamqp/rmqamqp_topologytransformer.t.cpp +++ b/src/tests/rmqamqp/rmqamqp_topologytransformer.t.cpp @@ -45,58 +45,41 @@ class TopologyTransformerTests : public ::testing::Test { void assertQueueDeclare(const Message& msg) { - ASSERT_TRUE(msg.is() && - msg.the().is() && - msg.the() - .the() - .is()); + ASSERT_TRUE((MessageUtil::isMethod(msg))); } void assertExchangeDeclare(const Message& msg) { - ASSERT_TRUE( - msg.is() && - msg.the().is() && - msg.the() - .the() - .is()); + ASSERT_TRUE((MessageUtil::isMethod(msg))); } void assertQueueBind(const Message& msg) { - ASSERT_TRUE(msg.is() && - msg.the().is() && - msg.the() - .the() - .is()); + ASSERT_TRUE( + (MessageUtil::isMethod( + msg))); } void assertQueueUnbind(const Message& msg) { - ASSERT_TRUE(msg.is() && - msg.the().is() && - msg.the() - .the() - .is()); + ASSERT_TRUE(( + MessageUtil::isMethod( + msg))); } void assertQueueDelete(const Message& msg) { - ASSERT_TRUE(msg.is() && - msg.the().is() && - msg.the() - .the() - .is()); + ASSERT_TRUE(( + MessageUtil::isMethod( + msg))); } void assertExchangeBind(const Message& msg) { - ASSERT_TRUE( - msg.is() && - msg.the().is() && - msg.the() - .the() - .is()); + ASSERT_TRUE((MessageUtil::isMethod(msg))); } };