Add Application Callback for Subscription Handling

src/app/InteractionModelEngine.cpp

@@ -331,6 +331,11 @@ CHIP_ERROR InteractionModelEngine::OnReadInitialRequest(Messaging::ExchangeConte
     ReadHandler * handler = mReadHandlers.CreateObject(*this, apExchangeContext, aInteractionType);
     if (handler)
     {
+        if (mpReadHandlerApplicationCallback)
+        {
+            handler->RegisterAppCallback(mpReadHandlerApplicationCallback);
+        }
+
         ReturnErrorOnFailure(handler->OnInitialRequest(std::move(aPayload)));
 
         aStatus = Protocols::InteractionModel::Status::Success;

src/app/InteractionModelEngine.h

@@ -65,7 +65,9 @@ namespace app {
  * handlers
  *
  */
-class InteractionModelEngine : public Messaging::ExchangeDelegate, public CommandHandler::Callback, public ReadHandler::Callback
+class InteractionModelEngine : public Messaging::ExchangeDelegate,
+                               public CommandHandler::Callback,
+                               public ReadHandler::ManagementCallback
 {
 public:
     /**
@@ -142,6 +144,15 @@ class InteractionModelEngine : public Messaging::ExchangeDelegate, public Comman
     CommandHandlerInterface * FindCommandHandler(EndpointId endpointId, ClusterId clusterId);
     void UnregisterCommandHandlers(EndpointId endpointId);
 
+    /*
+     * Register an application callback to be notified of notable events when handling reads/subscribes.
+     */
+    void RegisterReadHandlerAppCallback(ReadHandler::ApplicationCallback * mpApplicationCallback)
+    {
+        mpReadHandlerApplicationCallback = mpApplicationCallback;
+    }
+    void UnregisterReadHandlerAppCallback() { mpReadHandlerApplicationCallback = nullptr; }
+
     /**
      * Called when a timed interaction has failed (i.e. the exchange it was
      * happening on has closed while the exchange delegate was the timed
@@ -304,6 +315,8 @@ class InteractionModelEngine : public Messaging::ExchangeDelegate, public Comman
 
     ReadClient * mpActiveReadClientList = nullptr;
 
+    ReadHandler::ApplicationCallback * mpReadHandlerApplicationCallback = nullptr;
+
 #if CONFIG_IM_BUILD_FOR_UNIT_TEST
     int mReadHandlerCapacityOverride = -1;
 #endif

src/app/ReadHandler.cpp

@@ -36,7 +36,8 @@
 namespace chip {
 namespace app {
 
-ReadHandler::ReadHandler(Callback & apCallback, Messaging::ExchangeContext * apExchangeContext, InteractionType aInteractionType) :
+ReadHandler::ReadHandler(ManagementCallback & apCallback, Messaging::ExchangeContext * apExchangeContext,
+                         InteractionType aInteractionType) :
     mCallback(apCallback)
 {
     mpExchangeMgr           = apExchangeContext->GetExchangeMgr();
@@ -80,6 +81,11 @@ void ReadHandler::Abort(bool aCalledFromDestructor)
 
 ReadHandler::~ReadHandler()
 {
+    if (mActiveSubscription && mpApplicationCallback)
+    {
+        mpApplicationCallback->OnSubscriptionTerminated(*this);
+    }
+
     Abort(true);
 
     if (IsType(InteractionType::Subscribe))
@@ -159,10 +165,17 @@ CHIP_ERROR ReadHandler::OnStatusResponse(Messaging::ExchangeContext * apExchange
         {
             if (IsPriming())
             {
-                err           = SendSubscribeResponse();
+                err = SendSubscribeResponse();
+
                 mpExchangeCtx = nullptr;
                 SuccessOrExit(err);
+
                 mActiveSubscription = true;
+
+                if (mpApplicationCallback)
+                {
+                    mpApplicationCallback->OnSubscriptionEstablished(*this);
+                }
             }
             else
             {
@@ -690,6 +703,23 @@ CHIP_ERROR ReadHandler::ProcessSubscribeRequest(System::PacketBufferHandle && aP
     ReturnErrorOnFailure(subscribeRequestParser.GetMinIntervalFloorSeconds(&mMinIntervalFloorSeconds));
     ReturnErrorOnFailure(subscribeRequestParser.GetMaxIntervalCeilingSeconds(&mMaxIntervalCeilingSeconds));
     VerifyOrReturnError(mMinIntervalFloorSeconds <= mMaxIntervalCeilingSeconds, CHIP_ERROR_INVALID_ARGUMENT);
+
+    //
+    // Consult the application (if requested) of the impending subscription and whether we should still proceed to set it up.
+    // This also provides the application an opportunity to modify the negotiated min/max intervals set above.
+    //
+    if (mpApplicationCallback)
+    {
+        if (mpApplicationCallback->OnSubscriptionRequested(*this, *mpExchangeCtx->GetSessionHandle()->AsSecureSession()) !=
+            CHIP_NO_ERROR)
+        {
+            return CHIP_ERROR_TRANSACTION_CANCELED;
+        }
+    }
+
+    ChipLogProgress(DataManagement, "Final negotiated min/max parameters: Min = %ds, Max = %ds", mMinIntervalFloorSeconds,
+                    mMaxIntervalCeilingSeconds);
+
     ReturnErrorOnFailure(subscribeRequestParser.GetIsFabricFiltered(&mIsFabricFiltered));
     ReturnErrorOnFailure(Crypto::DRBG_get_bytes(reinterpret_cast<uint8_t *>(&mSubscriptionId), sizeof(mSubscriptionId)));
     ReturnErrorOnFailure(subscribeRequestParser.ExitContainer());

src/app/ReadHandler.h

@@ -51,7 +51,10 @@ namespace app {
 //
 namespace reporting {
 class Engine;
-}
+class TestReportingEngine;
+} // namespace reporting
+
+class InteractionModelEngine;
 
 /**
  *  @class ReadHandler
@@ -71,10 +74,13 @@ class ReadHandler : public Messaging::ExchangeDelegate
         Subscribe,
     };
 
-    class Callback
+    /*
+     * A callback used to manage the lifetime of the ReadHandler object.
+     */
+    class ManagementCallback
     {
     public:
-        virtual ~Callback() = default;
+        virtual ~ManagementCallback() = default;
 
         /*
          * Method that signals to a registered callback that this object
@@ -83,22 +89,96 @@ class ReadHandler : public Messaging::ExchangeDelegate
         virtual void OnDone(ReadHandler & apReadHandlerObj) = 0;
     };
 
+    /*
+     * A callback used to interact with the application.
+     */
+    class ApplicationCallback
+    {
+    public:
+        virtual ~ApplicationCallback() = default;
+
+        /*
+         * Called right after a SubscribeRequest has been parsed and processed. This notifies an interested application
+         * of a subscription that is about to be established. It also provides an avenue for altering the parameters of the
+         * subscription (specifically, the min/max negotiated intervals) or even outright rejecting the subscription for
+         * application-specific reasons.
+         *
+         * TODO: Need a new IM status code to convey application-rejected subscribes. Currently, an OUT_OF_RESOURCE error is sent
+         * back, which isn't appropriate
+         *
+         * To reject the subscription, a CHIP_ERROR code that is not equivalent to CHIP_NO_ERROR should be returned.
+         *
+         * More information about the set of paths associated with this subscription can be retrieved by calling the appropriate
+         * Get* methods below.
+         *
+         * aReadHandler:            Reference to the ReadHandler associated with the subscription.
+         * aSecureSession:          A reference to the underlying secure session associated with the subscription.
+         *
+         */
+        virtual CHIP_ERROR OnSubscriptionRequested(ReadHandler & aReadHandler, Transport::SecureSession & aSecureSession)
+        {
+            return CHIP_NO_ERROR;
+        }
+
+        /*
+         * Called after a subscription has been fully established.
+         */
+        virtual void OnSubscriptionEstablished(ReadHandler & aReadHandler){};
+
+        /*
+         * Called right before a subscription is about to get terminated. This is only called on subscriptions that were terminated
+         * after they had been fully established. This criteria includes a successful response to a call to
+         * OnSubscriptionEstablishment().
+         */
+        virtual void OnSubscriptionTerminated(ReadHandler & aReadHandler){};
+    };
+
+    /*
+     * Destructor - as part of destruction, it will abort the exchange context
+     * if a valid one still exists.
+     *
+     * See Abort() for details on when that might occur.
+     */
+    ~ReadHandler() override;
+
     /**
      *
      *  Constructor.
      *
      *  The callback passed in has to outlive this handler object.
      *
      */
-    ReadHandler(Callback & apCallback, Messaging::ExchangeContext * apExchangeContext, InteractionType aInteractionType);
+    ReadHandler(ManagementCallback & apCallback, Messaging::ExchangeContext * apExchangeContext, InteractionType aInteractionType);
+
+    ClusterInfo * GetAttributeClusterInfolist() { return mpAttributeClusterInfoList; }
+    ClusterInfo * GetEventClusterInfolist() { return mpEventClusterInfoList; }
+    ClusterInfo * GetDataVersionFilterlist() const { return mpDataVersionFilterList; }
+    EventNumber & GetEventMin() { return mEventMin; }
+    PriorityLevel GetCurrentPriority() { return mCurrentPriority; }
+
+    void GetReportingIntervals(uint16_t & aMinInterval, uint16_t & aMaxInterval)
+    {
+        aMinInterval = mMinIntervalFloorSeconds;
+        aMaxInterval = mMaxIntervalCeilingSeconds;
+    }
 
     /*
-     * Destructor - as part of destruction, it will abort the exchange context
-     * if a valid one still exists.
-     *
-     * See Abort() for details on when that might occur.
+     * Set the reporting intervals for the subscription. This SHALL only be called
+     * from the OnSubscriptionRequested callback above.
      */
-    ~ReadHandler() override;
+    CHIP_ERROR SetReportingIntervals(uint16_t aMinInterval, uint16_t aMaxInterval)
+    {
+        VerifyOrReturnError(IsGeneratingReports(), CHIP_ERROR_INCORRECT_STATE);
+        VerifyOrReturnError(aMinInterval <= aMaxInterval, CHIP_ERROR_INVALID_ARGUMENT);
+
+        mMinIntervalFloorSeconds   = aMinInterval;
+        mMaxIntervalCeilingSeconds = aMaxInterval;
+        return CHIP_NO_ERROR;
+    }
+
+private:
+    void RegisterAppCallback(ApplicationCallback * apCallback) { mpApplicationCallback = apCallback; }
+    void DeregisterAppCallback() { mpApplicationCallback = nullptr; }
 
     /**
      *  Process a read/subscribe request.  Parts of the processing may end up being asynchronous, but the ReadHandler
@@ -133,11 +213,6 @@ class ReadHandler : public Messaging::ExchangeDelegate
     bool IsAwaitingReportResponse() const { return mState == HandlerState::AwaitingReportResponse; }
 
     CHIP_ERROR ProcessDataVersionFilterList(DataVersionFilterIBs::Parser & aDataVersionFilterListParser);
-    ClusterInfo * GetAttributeClusterInfolist() { return mpAttributeClusterInfoList; }
-    ClusterInfo * GetEventClusterInfolist() { return mpEventClusterInfoList; }
-    ClusterInfo * GetDataVersionFilterlist() const { return mpDataVersionFilterList; }
-    EventNumber & GetEventMin() { return mEventMin; }
-    PriorityLevel GetCurrentPriority() { return mCurrentPriority; }
 
     // if current priority is in the middle, it has valid snapshoted last event number, it check cleaness via comparing
     // with snapshotted last event number. if current priority  is in the end, no valid
@@ -178,15 +253,16 @@ class ReadHandler : public Messaging::ExchangeDelegate
     uint32_t GetLastWrittenEventsBytes() const { return mLastWrittenEventsBytes; }
     CHIP_ERROR SendStatusReport(Protocols::InteractionModel::Status aStatus);
 
-private:
     friend class TestReadInteraction;
+    friend class chip::app::reporting::TestReportingEngine;
 
     //
     // The engine needs to be able to Abort/Close a ReadHandler instance upon completion of work for a given read/subscribe
     // interaction. We do not want to make these methods public just to give an adjacent class in the IM access, since public
     // should really be taking application usage considerations as well. Hence, make it a friend.
     //
     friend class chip::app::reporting::Engine;
+    friend class chip::app::InteractionModelEngine;
 
     enum class HandlerState
     {
@@ -252,7 +328,7 @@ class ReadHandler : public Messaging::ExchangeDelegate
     // The last schedule event number snapshoted in the beginning when preparing to fill new events to reports
     EventNumber mLastScheduledEventNumber      = 0;
     Messaging::ExchangeManager * mpExchangeMgr = nullptr;
-    Callback & mCallback;
+    ManagementCallback & mCallback;
 
     // Tracks whether we're in the initial phase of receiving priming
     // reports, which is always true for reads and true for subscriptions
@@ -285,6 +361,8 @@ class ReadHandler : public Messaging::ExchangeDelegate
     SubjectDescriptor mSubjectDescriptor;
     // The detailed encoding state for a single attribute, used by list chunking feature.
     AttributeValueEncoder::AttributeEncodeState mAttributeEncoderState;
+
+    ApplicationCallback * mpApplicationCallback = nullptr;
 };
 } // namespace app
 } // namespace chip

src/app/tests/TestReadInteraction.cpp

@@ -214,7 +214,7 @@ class MockInteractionModelApp : public chip::app::ReadClient::Callback
 // The typical callback implementor is the engine, but that would proceed to return the object
 // back to the handler pool (which we obviously don't want in this case). This just no-ops those calls.
 //
-class NullReadHandlerCallback : public chip::app::ReadHandler::Callback
+class NullReadHandlerCallback : public chip::app::ReadHandler::ManagementCallback
 {
 public:
     void OnDone(chip::app::ReadHandler & apReadHandlerObj) override {}

src/app/tests/TestReportingEngine.cpp

@@ -66,7 +66,7 @@ class TestExchangeDelegate : public Messaging::ExchangeDelegate
     void OnResponseTimeout(Messaging::ExchangeContext * ec) override {}
 };
 
-class DummyDelegate : public ReadHandler::Callback
+class DummyDelegate : public ReadHandler::ManagementCallback
 {
 public:
     void OnDone(ReadHandler & apHandler) override {}