[ReadHandler] Integration of ReportScheduler into the ReadHandler and IM engine.

src/app/InteractionModelEngine.cpp

@@ -51,16 +51,18 @@ InteractionModelEngine * InteractionModelEngine::GetInstance()
 }
 
 CHIP_ERROR InteractionModelEngine::Init(Messaging::ExchangeManager * apExchangeMgr, FabricTable * apFabricTable,
-                                        CASESessionManager * apCASESessionMgr,
+                                        reporting::ReportScheduler * reportScheduler, CASESessionManager * apCASESessionMgr,
                                         SubscriptionResumptionStorage * subscriptionResumptionStorage)
 {
     VerifyOrReturnError(apFabricTable != nullptr, CHIP_ERROR_INVALID_ARGUMENT);
     VerifyOrReturnError(apExchangeMgr != nullptr, CHIP_ERROR_INVALID_ARGUMENT);
+    VerifyOrReturnError(reportScheduler != nullptr, CHIP_ERROR_INVALID_ARGUMENT);
 
     mpExchangeMgr                   = apExchangeMgr;
     mpFabricTable                   = apFabricTable;
     mpCASESessionMgr                = apCASESessionMgr;
     mpSubscriptionResumptionStorage = subscriptionResumptionStorage;
+    mReportScheduler                = reportScheduler;
 
     ReturnErrorOnFailure(mpFabricTable->AddFabricDelegate(this));
     ReturnErrorOnFailure(mpExchangeMgr->RegisterUnsolicitedMessageHandlerForProtocol(Protocols::InteractionModel::Id, this));
@@ -741,7 +743,7 @@ Protocols::InteractionModel::Status InteractionModelEngine::OnReadInitialRequest
 
     // We have already reserved enough resources for read requests, and have granted enough resources for current subscriptions, so
     // we should be able to allocate resources requested by this request.
-    ReadHandler * handler = mReadHandlers.CreateObject(*this, apExchangeContext, aInteractionType);
+    ReadHandler * handler = mReadHandlers.CreateObject(*this, apExchangeContext, aInteractionType, mReportScheduler);
     if (handler == nullptr)
     {
         ChipLogProgress(InteractionModel, "no resource for %s interaction",
@@ -1845,7 +1847,7 @@ void InteractionModelEngine::ResumeSubscriptionsTimerCallback(System::Layer * ap
             return;
         }
 
-        ReadHandler * handler = imEngine->mReadHandlers.CreateObject(*imEngine);
+        ReadHandler * handler = imEngine->mReadHandlers.CreateObject(*imEngine, imEngine->GetReportScheduler());
         if (handler == nullptr)
         {
             ChipLogProgress(InteractionModel, "no resource for ReadHandler creation");

src/app/InteractionModelEngine.h

@@ -57,6 +57,7 @@
 #include <app/WriteClient.h>
 #include <app/WriteHandler.h>
 #include <app/reporting/Engine.h>
+#include <app/reporting/ReportScheduler.h>
 #include <app/util/attribute-metadata.h>
 #include <app/util/basic-types.h>
 
@@ -115,7 +116,7 @@ class InteractionModelEngine : public Messaging::UnsolicitedMessageHandler,
      *
      */
     CHIP_ERROR Init(Messaging::ExchangeManager * apExchangeMgr, FabricTable * apFabricTable,
-                    CASESessionManager * apCASESessionMgr                         = nullptr,
+                    reporting::ReportScheduler * reportScheduler, CASESessionManager * apCASESessionMgr = nullptr,
                     SubscriptionResumptionStorage * subscriptionResumptionStorage = nullptr);
 
     void Shutdown();
@@ -178,6 +179,8 @@ class InteractionModelEngine : public Messaging::UnsolicitedMessageHandler,
 
     reporting::Engine & GetReportingEngine() { return mReportingEngine; }
 
+    reporting::ReportScheduler * GetReportScheduler() { return mReportScheduler; }
+
     void ReleaseAttributePathList(ObjectList<AttributePathParams> *& aAttributePathList);
 
     CHIP_ERROR PushFrontAttributePathList(ObjectList<AttributePathParams> *& aAttributePathList,
@@ -566,6 +569,7 @@ class InteractionModelEngine : public Messaging::UnsolicitedMessageHandler,
     ObjectPool<TimedHandler, CHIP_IM_MAX_NUM_TIMED_HANDLER> mTimedHandlers;
     WriteHandler mWriteHandlers[CHIP_IM_MAX_NUM_WRITE_HANDLER];
     reporting::Engine mReportingEngine;
+    reporting::ReportScheduler * mReportScheduler = nullptr;
 
     static constexpr size_t kReservedHandlersForReads = kMinSupportedReadRequestsPerFabric * (CHIP_CONFIG_MAX_FABRICS);
     static constexpr size_t kReservedPathsForReads    = kMinSupportedPathsPerReadRequest * kReservedHandlersForReads;

src/app/ReadHandler.cpp

@@ -77,16 +77,9 @@ ReadHandler::ReadHandler(ManagementCallback & apCallback, Messaging::ExchangeCon
 
     mSessionHandle.Grab(mExchangeCtx->GetSessionHandle());
 
-// TODO (#27672): Uncomment when the ReportScheduler is implemented
-#if 0
-    if (nullptr != observer)
-    {
-        if (CHIP_NO_ERROR == SetObserver(observer))
-        {
-            mObserver->OnReadHandlerCreated(this);
-        }
-    }
-#endif
+    VerifyOrDie(observer != nullptr);
+    mObserver = observer;
+    mObserver->OnReadHandlerCreated(this);
 }
 
 #if CHIP_CONFIG_PERSIST_SUBSCRIPTIONS
@@ -97,16 +90,9 @@ ReadHandler::ReadHandler(ManagementCallback & apCallback, Observer * observer) :
     mInteractionType = InteractionType::Subscribe;
     mFlags.ClearAll();
 
-// TODO (#27672): Uncomment when the ReportScheduler is implemented
-#if 0
-    if (nullptr != observer)
-    {
-        if (CHIP_NO_ERROR == SetObserver(observer))
-        {
-            mObserver->OnReadHandlerCreated(this);
-        }
-    }
-#endif
+    VerifyOrDie(observer != nullptr);
+    mObserver = observer;
+    mObserver->OnReadHandlerCreated(this);
 }
 
 void ReadHandler::ResumeSubscription(CASESessionManager & caseSessionManager,
@@ -150,28 +136,14 @@ void ReadHandler::ResumeSubscription(CASESessionManager & caseSessionManager,
 
 ReadHandler::~ReadHandler()
 {
-    // TODO (#27672): Enable when the ReportScheduler is implemented and move in Close() after testing
-#if 0
-    if (nullptr != mObserver)
-    {
-        mObserver->OnReadHandlerDestroyed(this);
-    }
-#endif
+    mObserver->OnReadHandlerDestroyed(this);
+
     auto * appCallback = mManagementCallback.GetAppCallback();
     if (mFlags.Has(ReadHandlerFlags::ActiveSubscription) && appCallback)
     {
         appCallback->OnSubscriptionTerminated(*this);
     }
 
-    if (IsType(InteractionType::Subscribe))
-    {
-        InteractionModelEngine::GetInstance()->GetExchangeManager()->GetSessionManager()->SystemLayer()->CancelTimer(
-            MinIntervalExpiredCallback, this);
-
-        InteractionModelEngine::GetInstance()->GetExchangeManager()->GetSessionManager()->SystemLayer()->CancelTimer(
-            MaxIntervalExpiredCallback, this);
-    }
-
     if (IsAwaitingReportResponse())
     {
         InteractionModelEngine::GetInstance()->GetReportingEngine().OnReportConfirm();
@@ -290,7 +262,7 @@ CHIP_ERROR ReadHandler::OnStatusResponse(Messaging::ExchangeContext * apExchange
 
 CHIP_ERROR ReadHandler::SendStatusReport(Protocols::InteractionModel::Status aStatus)
 {
-    VerifyOrReturnLogError(IsReportableNow(), CHIP_ERROR_INCORRECT_STATE);
+    VerifyOrReturnLogError(mState == HandlerState::GeneratingReports, CHIP_ERROR_INCORRECT_STATE);
     if (IsPriming() || IsChunkedReport())
     {
         mSessionHandle.Grab(mExchangeCtx->GetSessionHandle());
@@ -314,7 +286,7 @@ CHIP_ERROR ReadHandler::SendStatusReport(Protocols::InteractionModel::Status aSt
 
 CHIP_ERROR ReadHandler::SendReportData(System::PacketBufferHandle && aPayload, bool aMoreChunks)
 {
-    VerifyOrReturnLogError(IsReportableNow(), CHIP_ERROR_INCORRECT_STATE);
+    VerifyOrReturnLogError(mState == HandlerState::GeneratingReports, CHIP_ERROR_INCORRECT_STATE);
     VerifyOrDie(!IsAwaitingReportResponse()); // Should not be reportable!
     if (IsPriming() || IsChunkedReport())
     {
@@ -359,21 +331,11 @@ CHIP_ERROR ReadHandler::SendReportData(System::PacketBufferHandle && aPayload, b
             InteractionModelEngine::GetInstance()->GetReportingEngine().OnReportConfirm();
         }
 
-        if (IsType(InteractionType::Subscribe) && !IsPriming())
+        // If we just finished a non-priming subscription report, notify our observers.
+        // Priming reports are handled when we send a SubscribeResponse.
+        if (IsType(InteractionType::Subscribe) && !IsPriming() && !IsChunkedReport())
         {
-// TODO (#27672): Enable when the ReportScheduler is implemented and remove call to UpdateReportTimer, will be handled by
-// the report Scheduler
-#if 0
-            if (nullptr != mObserver)
-            {
-                mObserver->OnSubscriptionAction(this);
-            }
-#endif
-
-            // Ignore the error from UpdateReportTimer.  If we've
-            // successfully sent the message, we need to return success from
-            // this method.
-            UpdateReportTimer();
+            mObserver->OnSubscriptionAction(this);
         }
     }
     if (!aMoreChunks)
@@ -641,16 +603,10 @@ void ReadHandler::MoveToState(const HandlerState aTargetState)
     // If we just unblocked sending reports, let's go ahead and schedule the reporting
     // engine to run to kick that off.
     //
-    if (aTargetState == HandlerState::GeneratingReports && IsReportableNow())
+    if (aTargetState == HandlerState::GeneratingReports)
     {
-// TODO (#27672): Enable when the ReportScheduler is implemented and remove the call to ScheduleRun()
-#if 0
-         if(nullptr != mObserver)
-        {
-            mObserver->OnBecameReportable(this);
-        }
-#endif
-        InteractionModelEngine::GetInstance()->GetReportingEngine().ScheduleRun();
+        // mObserver will take care of scheduling the report as soon as allowed
+        mObserver->OnBecameReportable(this);
     }
 }
 
@@ -691,15 +647,7 @@ CHIP_ERROR ReadHandler::SendSubscribeResponse()
     ReturnErrorOnFailure(writer.Finalize(&packet));
     VerifyOrReturnLogError(mExchangeCtx, CHIP_ERROR_INCORRECT_STATE);
 
-    // TODO (#27672): Uncomment when the ReportScheduler is implemented and remove call to UpdateReportTimer, handled by
-    // the report Scheduler
-#if 0
-    if (nullptr != mObserver)
-    {
-        mObserver->OnSubscriptionAction(this);
-    }
-#endif
-    ReturnErrorOnFailure(UpdateReportTimer());
+    mObserver->OnSubscriptionAction(this);
 
     ClearStateFlag(ReadHandlerFlags::PrimingReports);
     return mExchangeCtx->SendMessage(Protocols::InteractionModel::MsgType::SubscribeResponse, std::move(packet));
@@ -818,50 +766,6 @@ void ReadHandler::PersistSubscription()
     }
 }
 
-// TODO (#27672): Remove when ReportScheduler is enabled as timing will now be handled by the ReportScheduler
-void ReadHandler::MinIntervalExpiredCallback(System::Layer * apSystemLayer, void * apAppState)
-{
-    VerifyOrReturn(apAppState != nullptr);
-    ReadHandler * readHandler = static_cast<ReadHandler *>(apAppState);
-    ChipLogDetail(DataManagement, "Unblock report hold after min %d seconds", readHandler->mMinIntervalFloorSeconds);
-    readHandler->ClearStateFlag(ReadHandlerFlags::WaitingUntilMinInterval);
-    InteractionModelEngine::GetInstance()->GetExchangeManager()->GetSessionManager()->SystemLayer()->StartTimer(
-        System::Clock::Seconds16(readHandler->mMaxInterval - readHandler->mMinIntervalFloorSeconds), MaxIntervalExpiredCallback,
-        readHandler);
-}
-
-// TODO (#27672): Remove when ReportScheduler is enabled as timing will now be handled by the ReportScheduler
-void ReadHandler::MaxIntervalExpiredCallback(System::Layer * apSystemLayer, void * apAppState)
-{
-    VerifyOrReturn(apAppState != nullptr);
-    ReadHandler * readHandler = static_cast<ReadHandler *>(apAppState);
-    readHandler->ClearStateFlag(ReadHandlerFlags::WaitingUntilMaxInterval);
-    ChipLogProgress(DataManagement, "Refresh subscribe timer sync after %d seconds",
-                    readHandler->mMaxInterval - readHandler->mMinIntervalFloorSeconds);
-}
-
-// TODO (#27672): Remove when ReportScheduler is enabled as timing will now be handled by the ReportScheduler
-CHIP_ERROR ReadHandler::UpdateReportTimer()
-{
-    InteractionModelEngine::GetInstance()->GetExchangeManager()->GetSessionManager()->SystemLayer()->CancelTimer(
-        MinIntervalExpiredCallback, this);
-    InteractionModelEngine::GetInstance()->GetExchangeManager()->GetSessionManager()->SystemLayer()->CancelTimer(
-        MaxIntervalExpiredCallback, this);
-
-    if (!IsChunkedReport())
-    {
-        ChipLogProgress(DataManagement, "Refresh Subscribe Sync Timer with min %d seconds and max %d seconds",
-                        mMinIntervalFloorSeconds, mMaxInterval);
-        SetStateFlag(ReadHandlerFlags::WaitingUntilMinInterval);
-        SetStateFlag(ReadHandlerFlags::WaitingUntilMaxInterval);
-        ReturnErrorOnFailure(
-            InteractionModelEngine::GetInstance()->GetExchangeManager()->GetSessionManager()->SystemLayer()->StartTimer(
-                System::Clock::Seconds16(mMinIntervalFloorSeconds), MinIntervalExpiredCallback, this));
-    }
-
-    return CHIP_NO_ERROR;
-}
-
 void ReadHandler::ResetPathIterator()
 {
     mAttributePathExpandIterator = AttributePathExpandIterator(mpAttributePathList);
@@ -897,17 +801,8 @@ void ReadHandler::AttributePathIsDirty(const AttributePathParams & aAttributeCha
         mAttributeEncoderState = AttributeValueEncoder::AttributeEncodeState();
     }
 
-    if (IsReportableNow())
-    {
-        // TODO (#27672): Enable when the ReportScheduler is implemented and remove the call to ScheduleRun()
-#if 0
-        if(nullptr != mObserver)
-        {
-            mObserver->OnBecameReportable(this);
-        }
-#endif
-        InteractionModelEngine::GetInstance()->GetReportingEngine().ScheduleRun();
-    }
+    // ReportScheduler will take care of verifying the reportability of the handler and schedule the run
+    mObserver->OnBecameReportable(this);
 }
 
 Transport::SecureSession * ReadHandler::GetSession() const
@@ -926,20 +821,14 @@ void ReadHandler::ForceDirtyState()
 
 void ReadHandler::SetStateFlag(ReadHandlerFlags aFlag, bool aValue)
 {
-    bool oldReportable = IsReportableNow();
+    bool oldReportable = IsReportable();
     mFlags.Set(aFlag, aValue);
 
     // If we became reportable, schedule a reporting run.
-    if (!oldReportable && IsReportableNow())
+    if (!oldReportable && IsReportable())
     {
-// TODO (#27672): Enable when the ReportScheduler is implemented and remove the call to ScheduleRun()
-#if 0
-        if(nullptr != mObserver)
-        {
-            mObserver->OnBecameReportable(this);
-        }
-#endif
-        InteractionModelEngine::GetInstance()->GetReportingEngine().ScheduleRun();
+        // If we became reportable, the scheduler will schedule a run as soon as allowed
+        mObserver->OnBecameReportable(this);
     }
 }