Implement subscriptions spec

Based on [1] at 2963300.

pretty much all done by me, only a bit of dependency injection test
updates were done by cursor

for now don't test replaceData return value since not specified

Have not implemented RTL04b1's channel mode checking for same reason as
mentioned in 8d881e2.

[1] ably/specification#346

Author: lawrence-forooghian

CONTRIBUTING.md

@@ -44,7 +44,9 @@ To check formatting and code quality, run `swift run BuildTool lint`. Run with `
 ### Throwing errors
 
 - The public API of the SDK should use typed throws, and the thrown errors should be of type `ARTErrorInfo`.
-- `Dictionary.mapValues` does not support typed throws. We have our own extension `ablyLiveObjects_mapValuesWithTypedThrow` which does; use this.
+- Some platform methods do not support typed throws. In these cases, we have our own extension which does; use this instead. They are:
+  - `Dictionary.mapValues`; use `ablyLiveObjects_mapValuesWithTypedThrow`.
+  - `NSLock.withLock`; use `ablyLiveObjects_mapValuesWithTypedThrow`.
 
 ### Testing guidelines
 

Sources/AblyLiveObjects/DefaultRealtimeObjects.swift

@@ -138,6 +138,7 @@ internal final class DefaultRealtimeObjects: Sendable, LiveMapObjectPoolDelegate
         notYetImplemented()
     }
 
+    @discardableResult
     internal func on(event _: ObjectsEvent, callback _: ObjectsEventCallback) -> any OnObjectsEventResponse {
         notYetImplemented()
     }

Sources/AblyLiveObjects/Internal/DefaultLiveCounter.swift

@@ -96,14 +96,29 @@ internal final class DefaultLiveCounter: Sendable {
         notYetImplemented()
     }
 
-    internal func subscribe(listener _: @escaping LiveObjectUpdateCallback<LiveCounterUpdate>) -> any SubscribeResponse {
-        notYetImplemented()
+    @discardableResult
+    internal func subscribe(listener: @escaping LiveObjectUpdateCallback<DefaultLiveCounterUpdate>, coreSDK: CoreSDK) throws(ARTErrorInfo) -> any SubscribeResponse {
+        try mutex.ablyLiveObjects_withLockWithTypedThrow { () throws(ARTErrorInfo) in
+            // swiftlint:disable:next trailing_closure
+            try mutableState.liveObject.subscribe(listener: listener, coreSDK: coreSDK, updateSelfLater: { [weak self] updater in
+                guard let self else {
+                    return
+                }
+
+                mutex.withLock {
+                    updater(&mutableState.liveObject)
+                }
+            })
+        }
     }
 
     internal func unsubscribeAll() {
-        notYetImplemented()
+        mutex.withLock {
+            mutableState.liveObject.unsubscribeAll()
+        }
     }
 
+    @discardableResult
     internal func on(event _: LiveObjectLifecycleEvent, callback _: @escaping LiveObjectLifecycleEventCallback) -> any OnLiveObjectLifecycleEventResponse {
         notYetImplemented()
     }
@@ -112,31 +127,42 @@ internal final class DefaultLiveCounter: Sendable {
         notYetImplemented()
     }
 
+    // MARK: - Emitting update from external sources
+
+    /// Emit an event from this `LiveCounter`.
+    ///
+    /// This is used to instruct this counter to emit updates during an `OBJECT_SYNC`.
+    internal func emit(_ update: LiveObjectUpdate<DefaultLiveCounterUpdate>) {
+        mutex.withLock {
+            mutableState.liveObject.emit(update, on: userCallbackQueue)
+        }
+    }
+
     // MARK: - Data manipulation
 
     /// Replaces the internal data of this counter with the provided ObjectState, per RTLC6.
-    internal func replaceData(using state: ObjectState) {
+    internal func replaceData(using state: ObjectState) -> LiveObjectUpdate<DefaultLiveCounterUpdate> {
         mutex.withLock {
             mutableState.replaceData(using: state)
         }
     }
 
     /// Test-only method to merge initial value from an ObjectOperation, per RTLC10.
-    internal func testsOnly_mergeInitialValue(from operation: ObjectOperation) {
+    internal func testsOnly_mergeInitialValue(from operation: ObjectOperation) -> LiveObjectUpdate<DefaultLiveCounterUpdate> {
         mutex.withLock {
             mutableState.mergeInitialValue(from: operation)
         }
     }
 
     /// Test-only method to apply a COUNTER_CREATE operation, per RTLC8.
-    internal func testsOnly_applyCounterCreateOperation(_ operation: ObjectOperation) {
+    internal func testsOnly_applyCounterCreateOperation(_ operation: ObjectOperation) -> LiveObjectUpdate<DefaultLiveCounterUpdate> {
         mutex.withLock {
             mutableState.applyCounterCreateOperation(operation, logger: logger)
         }
     }
 
     /// Test-only method to apply a COUNTER_INC operation, per RTLC9.
-    internal func testsOnly_applyCounterIncOperation(_ operation: WireObjectsCounterOp?) {
+    internal func testsOnly_applyCounterIncOperation(_ operation: WireObjectsCounterOp?) -> LiveObjectUpdate<DefaultLiveCounterUpdate> {
         mutex.withLock {
             mutableState.applyCounterIncOperation(operation)
         }
@@ -156,6 +182,7 @@ internal final class DefaultLiveCounter: Sendable {
                 objectMessageSiteCode: objectMessageSiteCode,
                 objectsPool: &objectsPool,
                 logger: logger,
+                userCallbackQueue: userCallbackQueue,
             )
         }
     }
@@ -164,13 +191,13 @@ internal final class DefaultLiveCounter: Sendable {
 
     private struct MutableState {
         /// The mutable state common to all LiveObjects.
-        internal var liveObject: LiveObjectMutableState
+        internal var liveObject: LiveObjectMutableState<DefaultLiveCounterUpdate>
 
         /// The internal data that this map holds, per RTLC3.
         internal var data: Double
 
         /// Replaces the internal data of this counter with the provided ObjectState, per RTLC6.
-        internal mutating func replaceData(using state: ObjectState) {
+        internal mutating func replaceData(using state: ObjectState) -> LiveObjectUpdate<DefaultLiveCounterUpdate> {
             // RTLC6a: Replace the private siteTimeserials with the value from ObjectState.siteTimeserials
             liveObject.siteTimeserials = state.siteTimeserials
 
@@ -181,19 +208,32 @@ internal final class DefaultLiveCounter: Sendable {
             data = state.counter?.count?.doubleValue ?? 0
 
             // RTLC6d: If ObjectState.createOp is present, merge the initial value into the LiveCounter as described in RTLC10
-            if let createOp = state.createOp {
+            return if let createOp = state.createOp {
                 mergeInitialValue(from: createOp)
+            } else {
+                // TODO: I assume this is what to do, clarify in https://github.com/ably/specification/pull/346/files#r2201363446
+                .noop
             }
         }
 
         /// Merges the initial value from an ObjectOperation into this LiveCounter, per RTLC10.
-        internal mutating func mergeInitialValue(from operation: ObjectOperation) {
+        internal mutating func mergeInitialValue(from operation: ObjectOperation) -> LiveObjectUpdate<DefaultLiveCounterUpdate> {
+            let update: LiveObjectUpdate<DefaultLiveCounterUpdate>
+
             // RTLC10a: Add ObjectOperation.counter.count to data, if it exists
             if let operationCount = operation.counter?.count?.doubleValue {
                 data += operationCount
+                // RTLC10c
+                update = .update(DefaultLiveCounterUpdate(amount: operationCount))
+            } else {
+                // RTLC10d
+                update = .noop
             }
+
             // RTLC10b: Set the private flag createOperationIsMerged to true
             liveObject.createOperationIsMerged = true
+
+            return update
         }
 
         /// Attempts to apply an operation from an inbound `ObjectMessage`, per RTLC7.
@@ -203,6 +243,7 @@ internal final class DefaultLiveCounter: Sendable {
             objectMessageSiteCode: String?,
             objectsPool: inout ObjectsPool,
             logger: Logger,
+            userCallbackQueue: DispatchQueue,
         ) {
             guard let applicableOperation = liveObject.canApplyOperation(objectMessageSerial: objectMessageSerial, objectMessageSiteCode: objectMessageSiteCode, logger: logger) else {
                 // RTLC7b
@@ -216,13 +257,17 @@ internal final class DefaultLiveCounter: Sendable {
             switch operation.action {
             case .known(.counterCreate):
                 // RTLC7d1
-                applyCounterCreateOperation(
+                let update = applyCounterCreateOperation(
                     operation,
                     logger: logger,
                 )
+                // RTLC7d1a
+                liveObject.emit(update, on: userCallbackQueue)
             case .known(.counterInc):
                 // RTLC7d2
-                applyCounterIncOperation(operation.counterOp)
+                let update = applyCounterIncOperation(operation.counterOp)
+                // RTLC7d2a
+                liveObject.emit(update, on: userCallbackQueue)
             default:
                 // RTLC7d3
                 logger.log("Operation \(operation) has unsupported action for LiveCounter; discarding", level: .warn)
@@ -233,25 +278,28 @@ internal final class DefaultLiveCounter: Sendable {
         internal mutating func applyCounterCreateOperation(
             _ operation: ObjectOperation,
             logger: Logger,
-        ) {
+        ) -> LiveObjectUpdate<DefaultLiveCounterUpdate> {
             if liveObject.createOperationIsMerged {
                 // RTLC8b
                 logger.log("Not applying COUNTER_CREATE because a COUNTER_CREATE has already been applied", level: .warn)
-                return
+                return .noop
             }
 
-            // RTLC8c
-            mergeInitialValue(from: operation)
+            // RTLC8c, RTLC8e
+            return mergeInitialValue(from: operation)
         }
 
         /// Applies a `COUNTER_INC` operation, per RTLC9.
-        internal mutating func applyCounterIncOperation(_ operation: WireObjectsCounterOp?) {
+        internal mutating func applyCounterIncOperation(_ operation: WireObjectsCounterOp?) -> LiveObjectUpdate<DefaultLiveCounterUpdate> {
             guard let operation else {
-                return
+                // RTL9e
+                return .noop
             }
 
-            // RTLC9b
-            data += operation.amount.doubleValue
+            // RTLC9b, RTLC9d
+            let amount = operation.amount.doubleValue
+            data += amount
+            return .update(DefaultLiveCounterUpdate(amount: amount))
         }
     }
 }

Sources/AblyLiveObjects/Internal/DefaultLiveCounterUpdate.swift

@@ -0,0 +1,3 @@
+internal struct DefaultLiveCounterUpdate: LiveCounterUpdate, Equatable {
+    internal var amount: Double
+}