[Darwin] Issue 26012 - MTRDevice should stream subscription reports (#29358)

* [Darwin] Issue 26012 - MTRDevice should stream subscription reports

* Changed implementation to do per-packet batching

* Remove test/redundant code

* Added unit test protocol comment for readability

* Address review comment

Author: jtung-apple

src/darwin/Framework/CHIP/MTRBaseSubscriptionCallback.h

@@ -84,6 +84,11 @@ class MTRBaseSubscriptionCallback : public chip::app::ClusterStateCache::Callbac
         // Ensure we release the ReadClient before we tear down anything else,
         // so it can call our OnDeallocatePaths properly.
         mReadClient = nullptr;
+
+        // Make sure the block isn't run after object destruction
+        if (mInterimReportBlock) {
+            dispatch_block_cancel(mInterimReportBlock);
+        }
     }
 
     chip::app::BufferedReadCallback & GetBufferedCallback() { return mBufferedReadAdapter; }
@@ -103,6 +108,10 @@ class MTRBaseSubscriptionCallback : public chip::app::ClusterStateCache::Callbac
     // be immediately followed by OnDone and we want to do the deletion there.
     void ReportError(CHIP_ERROR aError, bool aCancelSubscription = true);
 
+    // Called at attribute/event report time to queue a block to report on the Matter queue so that for multi-packet reports, this
+    // block is run and reports in batch. No-op if the block is already queued.
+    void QueueInterimReport();
+
 private:
     void OnReportBegin() override;
 
@@ -166,6 +175,7 @@ class MTRBaseSubscriptionCallback : public chip::app::ClusterStateCache::Callbac
     std::unique_ptr<chip::app::ClusterStateCache> mClusterStateCache;
     bool mHaveQueuedDeletion = false;
     OnDoneHandler _Nullable mOnDoneHandler = nil;
+    dispatch_block_t mInterimReportBlock = nil;
 };
 
 NS_ASSUME_NONNULL_END

src/darwin/Framework/CHIP/MTRBaseSubscriptionCallback.mm

@@ -46,13 +46,35 @@
     if (attributeCallback != nil && attributeReports.count) {
         attributeCallback(attributeReports);
     }
+
     if (eventCallback != nil && eventReports.count) {
         eventCallback(eventReports);
     }
 }
 
+void MTRBaseSubscriptionCallback::QueueInterimReport()
+{
+    if (mInterimReportBlock) {
+        return;
+    }
+
+    mInterimReportBlock = dispatch_block_create(DISPATCH_BLOCK_INHERIT_QOS_CLASS, ^{
+        mInterimReportBlock = nil;
+        ReportData();
+        // Allocate reports arrays to continue accumulation
+        mAttributeReports = [NSMutableArray new];
+        mEventReports = [NSMutableArray new];
+    });
+
+    dispatch_async(DeviceLayer::PlatformMgrImpl().GetWorkQueue(), mInterimReportBlock);
+}
+
 void MTRBaseSubscriptionCallback::OnReportEnd()
 {
+    if (mInterimReportBlock) {
+        dispatch_block_cancel(mInterimReportBlock);
+        mInterimReportBlock = nil;
+    }
     ReportData();
     if (mReportEndHandler) {
         mReportEndHandler();

src/darwin/Framework/CHIP/MTRDevice.mm

@@ -187,6 +187,13 @@ @interface MTRDevice ()
 
 @end
 
+// Declaring selector so compiler won't complain about testing and calling it in _handleReportEnd
+#ifdef DEBUG
+@protocol MTRDeviceUnitTestDelegate <MTRDeviceDelegate>
+- (void)unitTestReportEndForDevice:(MTRDevice *)device;
+@end
+#endif
+
 @implementation MTRDevice
 
 - (instancetype)initWithNodeID:(NSNumber *)nodeID controller:(MTRDeviceController *)controller
@@ -402,9 +409,11 @@ - (void)_handleUnsolicitedMessageFromPublisher
     [self _changeState:MTRDeviceStateReachable];
 
     id<MTRDeviceDelegate> delegate = _weakDelegate.strongObject;
-    if (delegate && [delegate respondsToSelector:@selector(deviceBecameActive:)]) {
+    if (delegate) {
         dispatch_async(_delegateQueue, ^{
-            [delegate deviceBecameActive:self];
+            if ([delegate respondsToSelector:@selector(deviceBecameActive:)]) {
+                [delegate deviceBecameActive:self];
+            }
         });
     }
 
@@ -429,6 +438,17 @@ - (void)_handleReportEnd
 {
     os_unfair_lock_lock(&self->_lock);
     _estimatedStartTimeFromGeneralDiagnosticsUpTime = nil;
+// For unit testing only
+#ifdef DEBUG
+    id delegate = _weakDelegate.strongObject;
+    if (delegate) {
+        dispatch_async(_delegateQueue, ^{
+            if ([delegate respondsToSelector:@selector(unitTestReportEndForDevice:)]) {
+                [delegate unitTestReportEndForDevice:self];
+            }
+        });
+    }
+#endif
     os_unfair_lock_unlock(&self->_lock);
 }
 
@@ -1546,6 +1566,8 @@ - (void)invokeCommandWithEndpointID:(NSNumber *)endpointID
             [mEventReports addObject:[MTRBaseDevice eventReportForHeader:aEventHeader andData:value]];
         }
     }
+
+    QueueInterimReport();
 }
 
 void SubscriptionCallback::OnAttributeData(
@@ -1582,5 +1604,7 @@ - (void)invokeCommandWithEndpointID:(NSNumber *)endpointID
             [mAttributeReports addObject:@ { MTRAttributePathKey : attributePath, MTRDataKey : value }];
         }
     }
+
+    QueueInterimReport();
 }
 } // anonymous namespace

src/darwin/Framework/CHIPTests/MTRDeviceTests.m

@@ -122,6 +122,7 @@ @interface MTRDeviceTestDelegate : NSObject <MTRDeviceDelegate>
 @property (nonatomic, nullable) dispatch_block_t onNotReachable;
 @property (nonatomic, nullable) MTRDeviceTestDelegateDataHandler onAttributeDataReceived;
 @property (nonatomic, nullable) MTRDeviceTestDelegateDataHandler onEventDataReceived;
+@property (nonatomic, nullable) dispatch_block_t onReportEnd;
 @end
 
 @implementation MTRDeviceTestDelegate
@@ -148,6 +149,13 @@ - (void)device:(MTRDevice *)device receivedEventReport:(NSArray<NSDictionary<NSS
     }
 }
 
+- (void)unitTestReportEndForDevice:(MTRDevice *)device
+{
+    if (self.onReportEnd != nil) {
+        self.onReportEnd();
+    }
+}
+
 @end
 
 @interface MTRDeviceTests : XCTestCase
@@ -1457,6 +1465,8 @@ - (void)test017_TestMTRDeviceBasics
                 XCTAssertNotNil(eventDict[MTREventTimestampDateKey]);
             }
         }
+    };
+    delegate.onReportEnd = ^() {
         [gotReportsExpectation fulfill];
     };
 
@@ -1490,12 +1500,11 @@ - (void)test017_TestMTRDeviceBasics
 
     [self waitForExpectations:@[ subscriptionExpectation, gotReportsExpectation ] timeout:60];
 
+    delegate.onReportEnd = nil;
+
     XCTAssertNotEqual(attributeReportsReceived, 0);
     XCTAssertNotEqual(eventReportsReceived, 0);
 
-    attributeReportsReceived = 0;
-    eventReportsReceived = 0;
-
     // Before resubscribe, first test write failure and expected value effects
     NSNumber * testEndpointID = @(1);
     NSNumber * testClusterID = @(8);
@@ -1555,6 +1564,8 @@ - (void)test017_TestMTRDeviceBasics
     };
 
     // reset the onAttributeDataReceived to validate the following resubscribe test
+    attributeReportsReceived = 0;
+    eventReportsReceived = 0;
     delegate.onAttributeDataReceived = ^(NSArray<NSDictionary<NSString *, id> *> * data) {
         attributeReportsReceived += data.count;
     };