Selectively skip received methods after sending channel.close

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<rmqamqpt::ChannelMethod, rmqamqpt::ChannelClose>(
+            message) &&
+        !MessageUtil::isMethod<rmqamqpt::ChannelMethod,
+                               rmqamqpt::ChannelCloseOk>(message)) {
+        BALL_LOG_DEBUG << "Discarding frame received after sending "
+                          "Channel.Close";
+        return KEEP;
+    }
+
     if (message.is<rmqamqpt::Method>()) {
         const rmqamqpt::Method& method = message.the<rmqamqpt::Method>();
         switch (method.classId()) {

src/rmq/rmqamqp/rmqamqp_message.h

@@ -38,6 +38,23 @@ namespace rmqamqp {
 typedef bdlb::Variant<rmqamqpt::Heartbeat, rmqt::Message, rmqamqpt::Method>
     Message;
 
+struct MessageUtil {
+    template <typename MethodClassT, typename MethodTypeT>
+    static bool isMethod(const Message& message);
+};
+
+template <typename MethodClassT, typename MethodTypeT>
+bool MessageUtil::isMethod(const Message& message)
+{
+    if (!message.is<rmqamqpt::Method>()) {
+        return false;
+    }
+
+    const rmqamqpt::Method& method = message.the<rmqamqpt::Method>();
+    return method.is<MethodClassT>() &&
+           method.the<MethodClassT>().template is<MethodTypeT>();
+}
+
 } // namespace rmqamqp
 } // namespace BloombergLP
 

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<rmqamqpt::Method>()) {
-            return msg.the<rmqamqpt::Method>();
+        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<rmqamqpt::ConnectionMethod,
+                                           rmqamqpt::ConnectionCloseOk>(
+                decodeMessage(*frames[0]));
 
         boost::asio::post(d_eventLoop, callback);
 

src/tests/rmqamqp/rmqamqp_receivechannel.t.cpp

@@ -69,15 +69,6 @@ const MethodTypeT& MethodGetter(const bsl::shared_ptr<rmqamqp::Message>& msg)
     throw bsl::runtime_error("MethodMatcher: No Match");
 }
 
-template <typename MethodClassT, typename MethodTypeT>
-bool MethodMatcher(const bsl::shared_ptr<rmqamqp::Message>& msg)
-{
-    return (msg->is<rmqamqpt::Method>() &&
-            msg->the<rmqamqpt::Method>().is<MethodClassT>() &&
-            msg->the<rmqamqpt::Method>()
-                .the<MethodClassT>()
-                .template is<MethodTypeT>());
-}
 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> 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> receiveChannel = makeReceiveChannel(1);

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<rmqamqpt::Method>() &&
-                    msg.the<rmqamqpt::Method>().is<rmqamqpt::QueueMethod>() &&
-                    msg.the<rmqamqpt::Method>()
-                        .the<rmqamqpt::QueueMethod>()
-                        .is<rmqamqpt::QueueDeclare>());
+        ASSERT_TRUE((MessageUtil::isMethod<rmqamqpt::QueueMethod,
+                                           rmqamqpt::QueueDeclare>(msg)));
     }
 
     void assertExchangeDeclare(const Message& msg)
     {
-        ASSERT_TRUE(
-            msg.is<rmqamqpt::Method>() &&
-            msg.the<rmqamqpt::Method>().is<rmqamqpt::ExchangeMethod>() &&
-            msg.the<rmqamqpt::Method>()
-                .the<rmqamqpt::ExchangeMethod>()
-                .is<rmqamqpt::ExchangeDeclare>());
+        ASSERT_TRUE((MessageUtil::isMethod<rmqamqpt::ExchangeMethod,
+                                           rmqamqpt::ExchangeDeclare>(msg)));
     }
 
     void assertQueueBind(const Message& msg)
     {
-        ASSERT_TRUE(msg.is<rmqamqpt::Method>() &&
-                    msg.the<rmqamqpt::Method>().is<rmqamqpt::QueueMethod>() &&
-                    msg.the<rmqamqpt::Method>()
-                        .the<rmqamqpt::QueueMethod>()
-                        .is<rmqamqpt::QueueBind>());
+        ASSERT_TRUE(
+            (MessageUtil::isMethod<rmqamqpt::QueueMethod, rmqamqpt::QueueBind>(
+                msg)));
     }
 
     void assertQueueUnbind(const Message& msg)
     {
-        ASSERT_TRUE(msg.is<rmqamqpt::Method>() &&
-                    msg.the<rmqamqpt::Method>().is<rmqamqpt::QueueMethod>() &&
-                    msg.the<rmqamqpt::Method>()
-                        .the<rmqamqpt::QueueMethod>()
-                        .is<rmqamqpt::QueueUnbind>());
+        ASSERT_TRUE((
+            MessageUtil::isMethod<rmqamqpt::QueueMethod, rmqamqpt::QueueUnbind>(
+                msg)));
     }
 
     void assertQueueDelete(const Message& msg)
     {
-        ASSERT_TRUE(msg.is<rmqamqpt::Method>() &&
-                    msg.the<rmqamqpt::Method>().is<rmqamqpt::QueueMethod>() &&
-                    msg.the<rmqamqpt::Method>()
-                        .the<rmqamqpt::QueueMethod>()
-                        .is<rmqamqpt::QueueDelete>());
+        ASSERT_TRUE((
+            MessageUtil::isMethod<rmqamqpt::QueueMethod, rmqamqpt::QueueDelete>(
+                msg)));
     }
 
     void assertExchangeBind(const Message& msg)
     {
-        ASSERT_TRUE(
-            msg.is<rmqamqpt::Method>() &&
-            msg.the<rmqamqpt::Method>().is<rmqamqpt::ExchangeMethod>() &&
-            msg.the<rmqamqpt::Method>()
-                .the<rmqamqpt::ExchangeMethod>()
-                .is<rmqamqpt::ExchangeBind>());
+        ASSERT_TRUE((MessageUtil::isMethod<rmqamqpt::ExchangeMethod,
+                                           rmqamqpt::ExchangeBind>(msg)));
     }
 };