Implement increment / decrement / set / delete

Based on [1] at cb11ba8. Code by me, tests by Cursor and tidied up by
me.

Channel mode checking and echoMessages check omitted as in b6b0216.

[1] ably/specification#353

Author: lawrence-forooghian

Sources/AblyLiveObjects/Internal/InternalDefaultLiveCounter.swift

@@ -89,12 +89,43 @@ internal final class InternalDefaultLiveCounter: Sendable {
         }
     }
 
-    internal func increment(amount _: Double) async throws(ARTErrorInfo) {
-        notYetImplemented()
+    internal func increment(amount: Double, coreSDK: CoreSDK) async throws(ARTErrorInfo) {
+        do throws(InternalError) {
+            // RTLC12c
+            do {
+                try coreSDK.validateChannelState(notIn: [.detached, .failed, .suspended], operationDescription: "RealtimeObjects.createMap")
+            } catch {
+                throw error.toInternalError()
+            }
+
+            // RTLC12e1
+            if !amount.isFinite {
+                throw LiveObjectsError.counterIncrementAmountInvalid(amount: amount).toARTErrorInfo().toInternalError()
+            }
+
+            let objectMessage = OutboundObjectMessage(
+                operation: .init(
+                    // RTLC12e2
+                    action: .known(.counterInc),
+                    // RTLC12e3
+                    objectId: objectID,
+                    counterOp: .init(
+                        // RTLC12e4
+                        amount: .init(value: amount),
+                    ),
+                ),
+            )
+
+            // RTLC12f
+            try await coreSDK.publish(objectMessages: [objectMessage])
+        } catch {
+            throw error.toARTErrorInfo()
+        }
     }
 
-    internal func decrement(amount _: Double) async throws(ARTErrorInfo) {
-        notYetImplemented()
+    internal func decrement(amount: Double, coreSDK: CoreSDK) async throws(ARTErrorInfo) {
+        // RTLC13b
+        try await increment(amount: -amount, coreSDK: coreSDK)
     }
 
     @discardableResult

Sources/AblyLiveObjects/Internal/InternalDefaultLiveMap.swift

@@ -173,12 +173,63 @@ internal final class InternalDefaultLiveMap: Sendable {
         try entries(coreSDK: coreSDK, delegate: delegate).map(\.value)
     }
 
-    internal func set(key _: String, value _: LiveMapValue) async throws(ARTErrorInfo) {
-        notYetImplemented()
+    internal func set(key: String, value: InternalLiveMapValue, coreSDK: CoreSDK) async throws(ARTErrorInfo) {
+        do throws(InternalError) {
+            // RTLM20c
+            do {
+                try coreSDK.validateChannelState(notIn: [.detached, .failed, .suspended], operationDescription: "RealtimeObjects.createMap")
+            } catch {
+                throw error.toInternalError()
+            }
+
+            let objectMessage = OutboundObjectMessage(
+                operation: .init(
+                    // RTLM20e2
+                    action: .known(.mapSet),
+                    // RTLM20e3
+                    objectId: objectID,
+                    mapOp: .init(
+                        // RTLM20e4
+                        key: key,
+                        // RTLM20e5
+                        data: value.toObjectData,
+                    ),
+                ),
+            )
+
+            try await coreSDK.publish(objectMessages: [objectMessage])
+        } catch {
+            throw error.toARTErrorInfo()
+        }
     }
 
-    internal func remove(key _: String) async throws(ARTErrorInfo) {
-        notYetImplemented()
+    internal func remove(key: String, coreSDK: CoreSDK) async throws(ARTErrorInfo) {
+        do throws(InternalError) {
+            // RTLM21c
+            do {
+                try coreSDK.validateChannelState(notIn: [.detached, .failed, .suspended], operationDescription: "RealtimeObjects.createMap")
+            } catch {
+                throw error.toInternalError()
+            }
+
+            let objectMessage = OutboundObjectMessage(
+                operation: .init(
+                    // RTLM21e2
+                    action: .known(.mapRemove),
+                    // RTLM21e3
+                    objectId: objectID,
+                    mapOp: .init(
+                        // RTLM21e4
+                        key: key,
+                    ),
+                ),
+            )
+
+            // RTLM21f
+            try await coreSDK.publish(objectMessages: [objectMessage])
+        } catch {
+            throw error.toARTErrorInfo()
+        }
     }
 
     @discardableResult

Sources/AblyLiveObjects/Protocol/ObjectMessage.swift

@@ -18,7 +18,7 @@ internal struct InboundObjectMessage {
 }
 
 /// An `ObjectMessage` to be sent in the `state` property of an `OBJECT` `ProtocolMessage`.
-internal struct OutboundObjectMessage {
+internal struct OutboundObjectMessage: Equatable {
     internal var id: String? // OM2a
     internal var clientId: String? // OM2b
     internal var connectionId: String?
@@ -44,7 +44,7 @@ internal struct PartialObjectOperation {
     internal var initialValue: String? // OOP3h
 }
 
-internal struct ObjectOperation {
+internal struct ObjectOperation: Equatable {
     internal var action: WireEnum<ObjectOperationAction> // OOP3a
     internal var objectId: String // OOP3b
     internal var mapOp: ObjectsMapOp? // OOP3c
@@ -55,7 +55,7 @@ internal struct ObjectOperation {
     internal var initialValue: String? // OOP3h
 }
 
-internal struct ObjectData {
+internal struct ObjectData: Equatable {
     internal var objectId: String? // OD2a
     internal var boolean: Bool? // OD2c
     internal var bytes: Data? // OD2d
@@ -64,24 +64,24 @@ internal struct ObjectData {
     internal var json: JSONObjectOrArray? // TODO: Needs specification (see https://github.com/ably/ably-cocoa-liveobjects-plugin/issues/46)
 }
 
-internal struct ObjectsMapOp {
+internal struct ObjectsMapOp: Equatable {
     internal var key: String // OMO2a
     internal var data: ObjectData? // OMO2b
 }
 
-internal struct ObjectsMapEntry {
+internal struct ObjectsMapEntry: Equatable {
     internal var tombstone: Bool? // OME2a
     internal var timeserial: String? // OME2b
     internal var data: ObjectData // OME2c
     internal var serialTimestamp: Date? // OME2d
 }
 
-internal struct ObjectsMap {
+internal struct ObjectsMap: Equatable {
     internal var semantics: WireEnum<ObjectsMapSemantics> // OMP3a
     internal var entries: [String: ObjectsMapEntry]? // OMP3b
 }
 
-internal struct ObjectState {
+internal struct ObjectState: Equatable {
     internal var objectId: String // OST2a
     internal var siteTimeserials: [String: String] // OST2b
     internal var tombstone: Bool // OST2c

Sources/AblyLiveObjects/Protocol/WireObjectMessage.swift

@@ -356,7 +356,7 @@ extension WireObjectsMapOp: WireObjectCodable {
     }
 }
 
-internal struct WireObjectsCounterOp {
+internal struct WireObjectsCounterOp: Equatable {
     internal var amount: NSNumber // OCO2a
 }
 
@@ -410,7 +410,7 @@ extension WireObjectsMap: WireObjectCodable {
     }
 }
 
-internal struct WireObjectsCounter {
+internal struct WireObjectsCounter: Equatable {
     internal var count: NSNumber? // OCN2a
 }
 

Sources/AblyLiveObjects/Public/Public Proxy Objects/PublicDefaultLiveCounter.swift

@@ -27,11 +27,11 @@ internal final class PublicDefaultLiveCounter: LiveCounter {
     }
 
     internal func increment(amount: Double) async throws(ARTErrorInfo) {
-        try await proxied.increment(amount: amount)
+        try await proxied.increment(amount: amount, coreSDK: coreSDK)
     }
 
     internal func decrement(amount: Double) async throws(ARTErrorInfo) {
-        try await proxied.decrement(amount: amount)
+        try await proxied.decrement(amount: amount, coreSDK: coreSDK)
     }
 
     internal func subscribe(listener: @escaping LiveObjectUpdateCallback<LiveCounterUpdate>) throws(ARTErrorInfo) -> any SubscribeResponse {

Sources/AblyLiveObjects/Public/Public Proxy Objects/PublicDefaultLiveMap.swift

@@ -76,11 +76,13 @@ internal final class PublicDefaultLiveMap: LiveMap {
     }
 
     internal func set(key: String, value: LiveMapValue) async throws(ARTErrorInfo) {
-        try await proxied.set(key: key, value: value)
+        let internalValue = InternalLiveMapValue(liveMapValue: value)
+
+        try await proxied.set(key: key, value: internalValue, coreSDK: coreSDK)
     }
 
     internal func remove(key: String) async throws(ARTErrorInfo) {
-        try await proxied.remove(key: key)
+        try await proxied.remove(key: key, coreSDK: coreSDK)
     }
 
     internal func subscribe(listener: @escaping LiveObjectUpdateCallback<LiveMapUpdate>) throws(ARTErrorInfo) -> any SubscribeResponse {

Tests/AblyLiveObjectsTests/InternalDefaultLiveCounterTests.swift

@@ -422,4 +422,120 @@ struct InternalDefaultLiveCounterTests {
             #expect(subscriberInvocations.isEmpty)
         }
     }
+
+    /// Tests for the `increment` method, covering RTLC12 specification points
+    struct IncrementTests {
+        // @spec RTLC12c
+        @Test(arguments: [.detached, .failed, .suspended] as [ARTRealtimeChannelState])
+        func throwsErrorForInvalidChannelState(channelState: ARTRealtimeChannelState) async throws {
+            let logger = TestLogger()
+            let counter = InternalDefaultLiveCounter.createZeroValued(objectID: "arbitrary", logger: logger, userCallbackQueue: .main, clock: MockSimpleClock())
+            let coreSDK = MockCoreSDK(channelState: channelState)
+
+            await #expect {
+                try await counter.increment(amount: 10, coreSDK: coreSDK)
+            } throws: { error in
+                guard let errorInfo = error as? ARTErrorInfo else {
+                    return false
+                }
+
+                return errorInfo.code == 90001 && errorInfo.statusCode == 400
+            }
+        }
+
+        // @spec RTLC12e1 - The only part that is relevant in Swift's type system is the finiteness check
+        @Test(arguments: [
+            Double.nan,
+            Double.infinity,
+            -Double.infinity,
+        ] as [Double])
+        func throwsErrorForInvalidAmount(amount: Double) async throws {
+            let logger = TestLogger()
+            let counter = InternalDefaultLiveCounter.createZeroValued(objectID: "arbitrary", logger: logger, userCallbackQueue: .main, clock: MockSimpleClock())
+            let coreSDK = MockCoreSDK(channelState: .attached)
+
+            await #expect {
+                try await counter.increment(amount: amount, coreSDK: coreSDK)
+            } throws: { error in
+                guard let errorInfo = error as? ARTErrorInfo else {
+                    return false
+                }
+
+                return errorInfo.code == 40003 && errorInfo.statusCode == 400
+            }
+        }
+
+        // @spec RTLC12e2
+        // @spec RTLC12e3
+        // @spec RTLC12e4
+        // @spec RTLC12f
+        func publishesCorrectObjectMessage() async throws {
+            let logger = TestLogger()
+            let counter = InternalDefaultLiveCounter.createZeroValued(objectID: "counter:test@123", logger: logger, userCallbackQueue: .main, clock: MockSimpleClock())
+            let coreSDK = MockCoreSDK(channelState: .attached)
+
+            var publishedMessages: [OutboundObjectMessage] = []
+            coreSDK.setPublishHandler { messages in
+                publishedMessages.append(contentsOf: messages)
+            }
+
+            try await counter.increment(amount: 10.5, coreSDK: coreSDK)
+
+            let expectedMessage = OutboundObjectMessage(
+                operation: ObjectOperation(
+                    // RTLC12e2
+                    action: .known(.counterInc),
+                    // RTLC12e3
+                    objectId: "counter:test@123",
+                    // RTLC12e4
+                    counterOp: WireObjectsCounterOp(amount: NSNumber(value: 10.5)),
+                ),
+            )
+            // RTLC12f
+            #expect(publishedMessages.count == 1)
+            #expect(publishedMessages[0] == expectedMessage)
+        }
+
+        @Test
+        func throwsErrorWhenPublishFails() async throws {
+            let logger = TestLogger()
+            let counter = InternalDefaultLiveCounter.createZeroValued(objectID: "counter:test@123", logger: logger, userCallbackQueue: .main, clock: MockSimpleClock())
+            let coreSDK = MockCoreSDK(channelState: .attached)
+
+            coreSDK.setPublishHandler { _ throws(InternalError) in
+                throw InternalError.other(.generic(NSError(domain: "test", code: 0, userInfo: [NSLocalizedDescriptionKey: "Publish failed"])))
+            }
+
+            await #expect {
+                try await counter.increment(amount: 10, coreSDK: coreSDK)
+            } throws: { error in
+                guard let errorInfo = error as? ARTErrorInfo else {
+                    return false
+                }
+                return errorInfo.message.contains("Publish failed")
+            }
+        }
+    }
+
+    /// Tests for the `decrement` method, covering RTLC13 specification points
+    struct DecrementTests {
+        // @spec RTLC13b
+        @Test
+        func isOppositeOfIncrement() async throws {
+            // This is just a smoke test; we assume that this just calls `increment`, which is tested elsewhere.
+            let counter = InternalDefaultLiveCounter.createZeroValued(objectID: "counter:test@123", logger: TestLogger(), userCallbackQueue: .main, clock: MockSimpleClock())
+            let coreSDK = MockCoreSDK(channelState: .attached)
+
+            var publishedMessages: [OutboundObjectMessage] = []
+            coreSDK.setPublishHandler { messages in
+                publishedMessages.append(contentsOf: messages)
+            }
+
+            try await counter.decrement(amount: 10.5, coreSDK: coreSDK)
+
+            // RTLC12f
+            #expect(publishedMessages.count == 1)
+            #expect(publishedMessages[0].operation?.counterOp?.amount == -10.5 /* i.e. assert the amount gets negated */ )
+        }
+    }
 }