Merge pull request #42 from ably/groundwork-for-deletions-and-writes

Groundwork for deletions and writes

Author: lawrence-forooghian

Sources/AblyLiveObjects/Internal/CoreSDK.swift

@@ -5,7 +5,8 @@ internal import AblyPlugin
 ///
 /// This provides us with a mockable interface to ably-cocoa, and it also allows internal components and their tests not to need to worry about some of the boring details of how we bridge Swift types to AblyPlugin's Objective-C API (i.e. boxing).
 internal protocol CoreSDK: AnyObject, Sendable {
-    func sendObject(objectMessages: [OutboundObjectMessage]) async throws(InternalError)
+    /// Implements the internal `#publish` method of RTO15.
+    func publish(objectMessages: [OutboundObjectMessage]) async throws(InternalError)
 
     /// Returns the current state of the Realtime channel that this wraps.
     var channelState: ARTRealtimeChannelState { get }
@@ -28,7 +29,8 @@ internal final class DefaultCoreSDK: CoreSDK {
 
     // MARK: - CoreSDK conformance
 
-    internal func sendObject(objectMessages: [OutboundObjectMessage]) async throws(InternalError) {
+    internal func publish(objectMessages: [OutboundObjectMessage]) async throws(InternalError) {
+        // TODO: Implement the full spec of RTO15 (https://github.com/ably/ably-cocoa-liveobjects-plugin/issues/47)
         try await DefaultInternalPlugin.sendObject(
             objectMessages: objectMessages,
             channel: channel,
@@ -40,3 +42,28 @@ internal final class DefaultCoreSDK: CoreSDK {
         channel.state
     }
 }
+
+// MARK: - Channel State Validation
+
+/// Extension on CoreSDK to provide channel state validation utilities.
+internal extension CoreSDK {
+    /// Validates that the channel is not in any of the specified invalid states.
+    ///
+    /// - Parameters:
+    ///   - invalidStates: Array of channel states that are considered invalid for the operation
+    ///   - operationDescription: A description of the operation being performed, used in error messages
+    /// - Throws: `ARTErrorInfo` with code 90001 and statusCode 400 if the channel is in any of the invalid states
+    func validateChannelState(
+        notIn invalidStates: [ARTRealtimeChannelState],
+        operationDescription: String,
+    ) throws(ARTErrorInfo) {
+        let currentChannelState = channelState
+        if invalidStates.contains(currentChannelState) {
+            throw LiveObjectsError.objectsOperationFailedInvalidChannelState(
+                operationDescription: operationDescription,
+                channelState: currentChannelState,
+            )
+            .toARTErrorInfo()
+        }
+    }
+}

Sources/AblyLiveObjects/Internal/DefaultInternalPlugin.swift

@@ -38,7 +38,7 @@ internal final class DefaultInternalPlugin: NSObject, AblyPlugin.LiveObjectsInte
         let callbackQueue = pluginAPI.callbackQueue(for: client)
 
         logger.log("LiveObjects.DefaultInternalPlugin received prepare(_:)", level: .debug)
-        let liveObjects = InternalDefaultRealtimeObjects(logger: logger, userCallbackQueue: callbackQueue)
+        let liveObjects = InternalDefaultRealtimeObjects(logger: logger, userCallbackQueue: callbackQueue, clock: DefaultSimpleClock())
         pluginAPI.setPluginDataValue(liveObjects, forKey: Self.pluginDataKey, channel: channel)
     }
 

Sources/AblyLiveObjects/Internal/InternalDefaultLiveCounter.swift

@@ -11,45 +11,43 @@ internal final class InternalDefaultLiveCounter: Sendable {
 
     internal var testsOnly_siteTimeserials: [String: String] {
         mutex.withLock {
-            mutableState.liveObject.siteTimeserials
+            mutableState.liveObjectMutableState.siteTimeserials
         }
     }
 
     internal var testsOnly_createOperationIsMerged: Bool {
         mutex.withLock {
-            mutableState.liveObject.createOperationIsMerged
-        }
-    }
-
-    internal var testsOnly_objectID: String {
-        mutex.withLock {
-            mutableState.liveObject.objectID
+            mutableState.liveObjectMutableState.createOperationIsMerged
         }
     }
 
     private let logger: AblyPlugin.Logger
     private let userCallbackQueue: DispatchQueue
+    private let clock: SimpleClock
 
     // MARK: - Initialization
 
     internal convenience init(
         testsOnly_data data: Double,
         objectID: String,
         logger: AblyPlugin.Logger,
-        userCallbackQueue: DispatchQueue
+        userCallbackQueue: DispatchQueue,
+        clock: SimpleClock
     ) {
-        self.init(data: data, objectID: objectID, logger: logger, userCallbackQueue: userCallbackQueue)
+        self.init(data: data, objectID: objectID, logger: logger, userCallbackQueue: userCallbackQueue, clock: clock)
     }
 
     private init(
         data: Double,
         objectID: String,
         logger: AblyPlugin.Logger,
-        userCallbackQueue: DispatchQueue
+        userCallbackQueue: DispatchQueue,
+        clock: SimpleClock
     ) {
-        mutableState = .init(liveObject: .init(objectID: objectID), data: data)
+        mutableState = .init(liveObjectMutableState: .init(objectID: objectID), data: data)
         self.logger = logger
         self.userCallbackQueue = userCallbackQueue
+        self.clock = clock
     }
 
     /// Creates a "zero-value LiveCounter", per RTLC4.
@@ -60,27 +58,30 @@ internal final class InternalDefaultLiveCounter: Sendable {
         objectID: String,
         logger: AblyPlugin.Logger,
         userCallbackQueue: DispatchQueue,
+        clock: SimpleClock,
     ) -> Self {
         .init(
             data: 0,
             objectID: objectID,
             logger: logger,
             userCallbackQueue: userCallbackQueue,
+            clock: clock,
         )
     }
 
+    // MARK: - Data access
+
+    internal var objectID: String {
+        mutex.withLock {
+            mutableState.liveObjectMutableState.objectID
+        }
+    }
+
     // MARK: - Internal methods that back LiveCounter conformance
 
     internal func value(coreSDK: CoreSDK) throws(ARTErrorInfo) -> Double {
         // RTLC5b: If the channel is in the DETACHED or FAILED state, the library should indicate an error with code 90001
-        let currentChannelState = coreSDK.channelState
-        if currentChannelState == .detached || currentChannelState == .failed {
-            throw LiveObjectsError.objectsOperationFailedInvalidChannelState(
-                operationDescription: "LiveCounter.value",
-                channelState: currentChannelState,
-            )
-            .toARTErrorInfo()
-        }
+        try coreSDK.validateChannelState(notIn: [.detached, .failed], operationDescription: "LiveCounter.value")
 
         return mutex.withLock {
             // RTLC5c
@@ -100,21 +101,21 @@ internal final class InternalDefaultLiveCounter: Sendable {
     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] action in
+            try mutableState.liveObjectMutableState.subscribe(listener: listener, coreSDK: coreSDK, updateSelfLater: { [weak self] action in
                 guard let self else {
                     return
                 }
 
                 mutex.withLock {
-                    action(&mutableState.liveObject)
+                    action(&mutableState.liveObjectMutableState)
                 }
             })
         }
     }
 
     internal func unsubscribeAll() {
         mutex.withLock {
-            mutableState.liveObject.unsubscribeAll()
+            mutableState.liveObjectMutableState.unsubscribeAll()
         }
     }
 
@@ -134,21 +135,27 @@ internal final class InternalDefaultLiveCounter: Sendable {
     /// 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)
+            mutableState.liveObjectMutableState.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) -> LiveObjectUpdate<DefaultLiveCounterUpdate> {
+    ///
+    /// - Parameters:
+    ///   - objectMessageSerialTimestamp: The `serialTimestamp` of the containing `ObjectMessage`. Used if we need to tombstone this counter.
+    internal func replaceData(
+        using state: ObjectState,
+        objectMessageSerialTimestamp: Date?,
+    ) -> LiveObjectUpdate<DefaultLiveCounterUpdate> {
         mutex.withLock {
-            mutableState.replaceData(using: state)
+            mutableState.replaceData(using: state, objectMessageSerialTimestamp: objectMessageSerialTimestamp)
         }
     }
 
-    /// Test-only method to merge initial value from an ObjectOperation, per RTLC10.
-    internal func testsOnly_mergeInitialValue(from operation: ObjectOperation) -> LiveObjectUpdate<DefaultLiveCounterUpdate> {
+    /// Merges the initial value from an ObjectOperation into this LiveCounter, per RTLC10.
+    internal func mergeInitialValue(from operation: ObjectOperation) -> LiveObjectUpdate<DefaultLiveCounterUpdate> {
         mutex.withLock {
             mutableState.mergeInitialValue(from: operation)
         }
@@ -173,13 +180,15 @@ internal final class InternalDefaultLiveCounter: Sendable {
         _ operation: ObjectOperation,
         objectMessageSerial: String?,
         objectMessageSiteCode: String?,
+        objectMessageSerialTimestamp: Date?,
         objectsPool: inout ObjectsPool,
     ) {
         mutex.withLock {
             mutableState.apply(
                 operation,
                 objectMessageSerial: objectMessageSerial,
                 objectMessageSiteCode: objectMessageSiteCode,
+                objectMessageSerialTimestamp: objectMessageSerialTimestamp,
                 objectsPool: &objectsPool,
                 logger: logger,
                 userCallbackQueue: userCallbackQueue,
@@ -189,20 +198,26 @@ internal final class InternalDefaultLiveCounter: Sendable {
 
     // MARK: - Mutable state and the operations that affect it
 
-    private struct MutableState {
+    private struct MutableState: InternalLiveObject {
         /// The mutable state common to all LiveObjects.
-        internal var liveObject: LiveObjectMutableState<DefaultLiveCounterUpdate>
+        internal var liveObjectMutableState: 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) -> LiveObjectUpdate<DefaultLiveCounterUpdate> {
+        ///
+        /// - Parameters:
+        ///   - objectMessageSerialTimestamp: The `serialTimestamp` of the containing `ObjectMessage`. Used if we need to tombstone this counter.
+        internal mutating func replaceData(
+            using state: ObjectState,
+            objectMessageSerialTimestamp: Date?,
+        ) -> LiveObjectUpdate<DefaultLiveCounterUpdate> {
             // RTLC6a: Replace the private siteTimeserials with the value from ObjectState.siteTimeserials
-            liveObject.siteTimeserials = state.siteTimeserials
+            liveObjectMutableState.siteTimeserials = state.siteTimeserials
 
             // RTLC6b: Set the private flag createOperationIsMerged to false
-            liveObject.createOperationIsMerged = false
+            liveObjectMutableState.createOperationIsMerged = false
 
             // RTLC6c: Set data to the value of ObjectState.counter.count, or to 0 if it does not exist
             data = state.counter?.count?.doubleValue ?? 0
@@ -231,7 +246,7 @@ internal final class InternalDefaultLiveCounter: Sendable {
             }
 
             // RTLC10b: Set the private flag createOperationIsMerged to true
-            liveObject.createOperationIsMerged = true
+            liveObjectMutableState.createOperationIsMerged = true
 
             return update
         }
@@ -241,18 +256,19 @@ internal final class InternalDefaultLiveCounter: Sendable {
             _ operation: ObjectOperation,
             objectMessageSerial: String?,
             objectMessageSiteCode: String?,
+            objectMessageSerialTimestamp: Date?,
             objectsPool: inout ObjectsPool,
             logger: Logger,
             userCallbackQueue: DispatchQueue,
         ) {
-            guard let applicableOperation = liveObject.canApplyOperation(objectMessageSerial: objectMessageSerial, objectMessageSiteCode: objectMessageSiteCode, logger: logger) else {
+            guard let applicableOperation = liveObjectMutableState.canApplyOperation(objectMessageSerial: objectMessageSerial, objectMessageSiteCode: objectMessageSiteCode, logger: logger) else {
                 // RTLC7b
                 logger.log("Operation \(operation) (serial: \(String(describing: objectMessageSerial)), siteCode: \(String(describing: objectMessageSiteCode))) should not be applied; discarding", level: .debug)
                 return
             }
 
             // RTLC7c
-            liveObject.siteTimeserials[applicableOperation.objectMessageSiteCode] = applicableOperation.objectMessageSerial
+            liveObjectMutableState.siteTimeserials[applicableOperation.objectMessageSiteCode] = applicableOperation.objectMessageSerial
 
             switch operation.action {
             case .known(.counterCreate):
@@ -262,12 +278,12 @@ internal final class InternalDefaultLiveCounter: Sendable {
                     logger: logger,
                 )
                 // RTLC7d1a
-                liveObject.emit(update, on: userCallbackQueue)
+                liveObjectMutableState.emit(update, on: userCallbackQueue)
             case .known(.counterInc):
                 // RTLC7d2
                 let update = applyCounterIncOperation(operation.counterOp)
                 // RTLC7d2a
-                liveObject.emit(update, on: userCallbackQueue)
+                liveObjectMutableState.emit(update, on: userCallbackQueue)
             default:
                 // RTLC7d3
                 logger.log("Operation \(operation) has unsupported action for LiveCounter; discarding", level: .warn)
@@ -279,7 +295,7 @@ internal final class InternalDefaultLiveCounter: Sendable {
             _ operation: ObjectOperation,
             logger: Logger,
         ) -> LiveObjectUpdate<DefaultLiveCounterUpdate> {
-            if liveObject.createOperationIsMerged {
+            if liveObjectMutableState.createOperationIsMerged {
                 // RTLC8b
                 logger.log("Not applying COUNTER_CREATE because a COUNTER_CREATE has already been applied", level: .warn)
                 return .noop