apply keepSubscription logic for subscription only (#13807)

Author: yunhanw-google

src/app/InteractionModelEngine.cpp

@@ -205,34 +205,50 @@ CHIP_ERROR InteractionModelEngine::OnInvokeCommandRequest(Messaging::ExchangeCon
     return CHIP_NO_ERROR;
 }
 
-CHIP_ERROR InteractionModelEngine::OnReadInitialRequest(Messaging::ExchangeContext * apExchangeContext,
-                                                        const PayloadHeader & aPayloadHeader,
-                                                        System::PacketBufferHandle && aPayload,
-                                                        ReadHandler::InteractionType aInteractionType,
-                                                        Protocols::InteractionModel::Status & aStatus)
+CHIP_ERROR InteractionModelEngine::ShutdownExistingSubscriptionsIfNeeded(Messaging::ExchangeContext * apExchangeContext,
+                                                                         System::PacketBufferHandle && aPayload)
 {
-    ChipLogDetail(InteractionModel, "Received %s request",
-                  aInteractionType == ReadHandler::InteractionType::Subscribe ? "Subscribe" : "Read");
+    bool keepSubscriptions = true;
+    System::PacketBufferTLVReader reader;
+    reader.Init(std::move(aPayload));
+    ReturnErrorOnFailure(reader.Next());
+    SubscribeRequestMessage::Parser subscribeRequestParser;
+    ReturnErrorOnFailure(subscribeRequestParser.Init(reader));
+    CHIP_ERROR err = subscribeRequestParser.GetKeepSubscriptions(&keepSubscriptions);
+    if (CHIP_END_OF_TLV == err)
+    {
+        return CHIP_NO_ERROR;
+    }
+    else if (CHIP_NO_ERROR != err)
+    {
+        return err;
+    }
+
+    if (keepSubscriptions)
+    {
+        return CHIP_NO_ERROR;
+    }
 
     for (auto & readHandler : mReadHandlers)
     {
         if (!readHandler.IsFree() && readHandler.IsSubscriptionType() &&
             readHandler.GetInitiatorNodeId() == apExchangeContext->GetSessionHandle()->AsSecureSession()->GetPeerNodeId() &&
             readHandler.GetAccessingFabricIndex() == apExchangeContext->GetSessionHandle()->AsSecureSession()->GetFabricIndex())
         {
-            bool keepSubscriptions = true;
-            System::PacketBufferTLVReader reader;
-            reader.Init(aPayload.Retain());
-            ReturnErrorOnFailure(reader.Next());
-            SubscribeRequestMessage::Parser subscribeRequestParser;
-            ReturnErrorOnFailure(subscribeRequestParser.Init(reader));
-            CHIP_ERROR err = subscribeRequestParser.GetKeepSubscriptions(&keepSubscriptions);
-            if (err == CHIP_NO_ERROR && !keepSubscriptions)
-            {
-                readHandler.Shutdown(ReadHandler::ShutdownOptions::AbortCurrentExchange);
-            }
+            readHandler.Shutdown(ReadHandler::ShutdownOptions::AbortCurrentExchange);
         }
     }
+    return CHIP_NO_ERROR;
+}
+
+CHIP_ERROR InteractionModelEngine::OnReadInitialRequest(Messaging::ExchangeContext * apExchangeContext,
+                                                        const PayloadHeader & aPayloadHeader,
+                                                        System::PacketBufferHandle && aPayload,
+                                                        ReadHandler::InteractionType aInteractionType,
+                                                        Protocols::InteractionModel::Status & aStatus)
+{
+    ChipLogDetail(InteractionModel, "Received %s request",
+                  aInteractionType == ReadHandler::InteractionType::Subscribe ? "Subscribe" : "Read");
 
     // Reserve the last ReadHandler for ReadInteraction
     if (aInteractionType == ReadHandler::InteractionType::Subscribe &&
@@ -355,7 +371,9 @@ CHIP_ERROR InteractionModelEngine::OnMessageReceived(Messaging::ExchangeContext
     }
     else if (aPayloadHeader.HasMessageType(Protocols::InteractionModel::MsgType::SubscribeRequest))
     {
-        SuccessOrExit(OnReadInitialRequest(apExchangeContext, aPayloadHeader, std::move(aPayload),
+        System::PacketBufferHandle payload = aPayload.Retain();
+        SuccessOrExit(ShutdownExistingSubscriptionsIfNeeded(apExchangeContext, std::move(aPayload)));
+        SuccessOrExit(OnReadInitialRequest(apExchangeContext, aPayloadHeader, std::move(payload),
                                            ReadHandler::InteractionType::Subscribe, status));
     }
     else if (aPayloadHeader.HasMessageType(Protocols::InteractionModel::MsgType::ReportData))

src/app/InteractionModelEngine.h

@@ -232,6 +232,9 @@ class InteractionModelEngine : public Messaging::ExchangeDelegate, public Comman
 
     bool HasActiveRead();
 
+    CHIP_ERROR ShutdownExistingSubscriptionsIfNeeded(Messaging::ExchangeContext * apExchangeContext,
+                                                     System::PacketBufferHandle && aPayload);
+
     Messaging::ExchangeManager * mpExchangeMgr = nullptr;
     InteractionModelDelegate * mpDelegate      = nullptr;