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 e65643a.

[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/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 */ )
+        }
+    }
 }