feat(DataStore): DisableSubscriptions flag for watchOS

AmplifyPlugins/DataStore/Sources/AWSDataStorePlugin/Configuration/DataStoreConfiguration+Helper.swift

@@ -15,6 +15,41 @@ extension DataStoreConfiguration {
     public static let defaultSyncMaxRecords: UInt = 10_000
     public static let defaultSyncPageSize: UInt = 1_000
 
+    #if os(watchOS)
+    /// Creates a custom configuration. The only required property is `conflictHandler`.
+    ///
+    /// - Parameters:
+    ///   - errorHandler: a callback function called on unhandled errors
+    ///   - conflictHandler: a callback called when a conflict could not be resolved by the service
+    ///   - syncInterval: how often the sync engine will run (in seconds)
+    ///   - syncMaxRecords: the number of records to sync per execution
+    ///   - syncPageSize: the page size of each sync execution
+    ///   - authModeStrategy: authorization strategy (.default | multiauth)
+    /// - Returns: an instance of `DataStoreConfiguration` with the passed parameters.
+    public static func custom(
+        errorHandler: @escaping DataStoreErrorHandler = { error in
+            Amplify.Logging.error(error: error)
+        },
+        conflictHandler: @escaping DataStoreConflictHandler = { _, resolve  in
+            resolve(.applyRemote)
+        },
+        syncInterval: TimeInterval = DataStoreConfiguration.defaultSyncInterval,
+        syncMaxRecords: UInt = DataStoreConfiguration.defaultSyncMaxRecords,
+        syncPageSize: UInt = DataStoreConfiguration.defaultSyncPageSize,
+        syncExpressions: [DataStoreSyncExpression] = [],
+        authModeStrategy: AuthModeStrategyType = .default,
+        disableSubscriptions: @escaping () -> Bool
+    ) -> DataStoreConfiguration {
+        return DataStoreConfiguration(errorHandler: errorHandler,
+                                      conflictHandler: conflictHandler,
+                                      syncInterval: syncInterval,
+                                      syncMaxRecords: syncMaxRecords,
+                                      syncPageSize: syncPageSize,
+                                      syncExpressions: syncExpressions,
+                                      authModeStrategy: authModeStrategy,
+                                      disableSubscriptions: disableSubscriptions)
+    }
+    #else
     /// Creates a custom configuration. The only required property is `conflictHandler`.
     ///
     /// - Parameters:
@@ -46,6 +81,7 @@ extension DataStoreConfiguration {
                                       syncExpressions: syncExpressions,
                                       authModeStrategy: authModeStrategy)
     }
+    #endif
 
     /// The default configuration.
     public static var `default`: DataStoreConfiguration {

AmplifyPlugins/DataStore/Sources/AWSDataStorePlugin/Configuration/DataStoreConfiguration.swift

@@ -70,6 +70,27 @@ public struct DataStoreConfiguration {
     /// Authorization mode strategy
     public var authModeStrategyType: AuthModeStrategyType
 
+    public let disableSubscriptions: () -> Bool
+
+    #if os(watchOS)
+    init(errorHandler: @escaping DataStoreErrorHandler,
+         conflictHandler: @escaping DataStoreConflictHandler,
+         syncInterval: TimeInterval,
+         syncMaxRecords: UInt,
+         syncPageSize: UInt,
+         syncExpressions: [DataStoreSyncExpression],
+         authModeStrategy: AuthModeStrategyType = .default,
+         disableSubscriptions: @escaping () -> Bool) {
+        self.errorHandler = errorHandler
+        self.conflictHandler = conflictHandler
+        self.syncInterval = syncInterval
+        self.syncMaxRecords = syncMaxRecords
+        self.syncPageSize = syncPageSize
+        self.syncExpressions = syncExpressions
+        self.authModeStrategyType = authModeStrategy
+        self.disableSubscriptions = disableSubscriptions
+    }
+    #else
     init(errorHandler: @escaping DataStoreErrorHandler,
          conflictHandler: @escaping DataStoreConflictHandler,
          syncInterval: TimeInterval,
@@ -84,6 +105,7 @@ public struct DataStoreConfiguration {
         self.syncPageSize = syncPageSize
         self.syncExpressions = syncExpressions
         self.authModeStrategyType = authModeStrategy
+        self.disableSubscriptions = { false }
     }
-
+    #endif
 }

AmplifyPlugins/DataStore/Sources/AWSDataStorePlugin/Sync/RemoteSyncEngine.swift

@@ -86,7 +86,7 @@ class RemoteSyncEngine: RemoteSyncEngineBehavior {
                                   authModeStrategy: resolvedAuthStrategy)
 
         let reconciliationQueueFactory = reconciliationQueueFactory ??
-            AWSIncomingEventReconciliationQueue.init(modelSchemas:api:storageAdapter:syncExpressions:auth:authModeStrategy:modelReconciliationQueueFactory:)
+        AWSIncomingEventReconciliationQueue.init(modelSchemas:api:storageAdapter:syncExpressions:auth:authModeStrategy:modelReconciliationQueueFactory:disableRealTimeUpdates:)
 
         let initialSyncOrchestratorFactory = initialSyncOrchestratorFactory ??
             AWSInitialSyncOrchestrator.init(dataStoreConfiguration:authModeStrategy:api:reconciliationQueue:storageAdapter:)
@@ -289,7 +289,8 @@ class RemoteSyncEngine: RemoteSyncEngineBehavior {
                                                                dataStoreConfiguration.syncExpressions,
                                                                auth,
                                                                authModeStrategy,
-                                                               nil)
+                                                               nil,
+                                                               dataStoreConfiguration.disableSubscriptions)
         reconciliationQueueSink = reconciliationQueue?
             .publisher
             .sink(

AmplifyPlugins/DataStore/Sources/AWSDataStorePlugin/Sync/SubscriptionSync/AWSIncomingEventReconciliationQueue.swift

@@ -10,6 +10,8 @@ import AWSPluginsCore
 import Combine
 import Foundation
 
+typealias DisableRealTimeUpdates = () -> Bool
+
 // Used for testing:
 typealias IncomingEventReconciliationQueueFactory =
     ([ModelSchema],
@@ -18,7 +20,8 @@ typealias IncomingEventReconciliationQueueFactory =
     [DataStoreSyncExpression],
     AuthCategoryBehavior?,
     AuthModeStrategy,
-    ModelReconciliationQueueFactory?
+    ModelReconciliationQueueFactory?,
+    @escaping DisableRealTimeUpdates
 ) async -> IncomingEventReconciliationQueue
 
 final class AWSIncomingEventReconciliationQueue: IncomingEventReconciliationQueue {
@@ -48,7 +51,8 @@ final class AWSIncomingEventReconciliationQueue: IncomingEventReconciliationQueu
          syncExpressions: [DataStoreSyncExpression],
          auth: AuthCategoryBehavior? = nil,
          authModeStrategy: AuthModeStrategy,
-         modelReconciliationQueueFactory: ModelReconciliationQueueFactory? = nil) async {
+         modelReconciliationQueueFactory: ModelReconciliationQueueFactory? = nil,
+         disableRealTimeUpdates: @escaping () -> Bool = { false } ) async {
         self.modelSchemasCount = modelSchemas.count
         self.modelReconciliationQueueSinks.set([:])
         self.eventReconciliationQueueTopic = CurrentValueSubject<IncomingEventReconciliationQueueEvent, DataStoreError>(.idle)
@@ -78,13 +82,13 @@ final class AWSIncomingEventReconciliationQueue: IncomingEventReconciliationQueu
                 continue
             }
             let queue = await self.modelReconciliationQueueFactory(modelSchema,
-                                                             storageAdapter,
-                                                             api,
-                                                             reconcileAndSaveQueue,
-                                                             modelPredicate,
-                                                             auth,
-                                                             authModeStrategy,
-                                                             nil)
+                                                                   storageAdapter,
+                                                                   api,
+                                                                   reconcileAndSaveQueue,
+                                                                   modelPredicate,
+                                                                   auth,
+                                                                   authModeStrategy,
+                                                                   disableRealTimeUpdates() ? OperationDisabledIncomingSubscriptionEventPublisher() : nil)
 
             reconciliationQueues.with { reconciliationQueues in
                 reconciliationQueues[modelName] = queue
@@ -190,14 +194,15 @@ extension AWSIncomingEventReconciliationQueue: DefaultLogger {
 
 // MARK: - Static factory
 extension AWSIncomingEventReconciliationQueue {
-    static let factory: IncomingEventReconciliationQueueFactory = { modelSchemas, api, storageAdapter, syncExpressions, auth, authModeStrategy, _ in
+    static let factory: IncomingEventReconciliationQueueFactory = { modelSchemas, api, storageAdapter, syncExpressions, auth, authModeStrategy, _, disableRealTimeUpdates in
         await AWSIncomingEventReconciliationQueue(modelSchemas: modelSchemas,
-                                            api: api,
-                                            storageAdapter: storageAdapter,
-                                            syncExpressions: syncExpressions,
-                                            auth: auth,
-                                            authModeStrategy: authModeStrategy,
-                                            modelReconciliationQueueFactory: nil)
+                                                  api: api,
+                                                  storageAdapter: storageAdapter,
+                                                  syncExpressions: syncExpressions,
+                                                  auth: auth,
+                                                  authModeStrategy: authModeStrategy,
+                                                  modelReconciliationQueueFactory: nil,
+                                                  disableRealTimeUpdates: disableRealTimeUpdates)
     }
 }
 

AmplifyPlugins/DataStore/Sources/AWSDataStorePlugin/Sync/SubscriptionSync/OperationDisabledSubscriptionEventPublisher.swift

@@ -0,0 +1,31 @@
+//
+// Copyright Amazon.com Inc. or its affiliates.
+// All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import Amplify
+import AWSPluginsCore
+import Combine
+
+final class OperationDisabledIncomingSubscriptionEventPublisher: IncomingSubscriptionEventPublisher {
+
+    private let subscriptionEventSubject: PassthroughSubject<IncomingSubscriptionEventPublisherEvent, DataStoreError>
+
+    var publisher: AnyPublisher<IncomingSubscriptionEventPublisherEvent, DataStoreError> {
+        return subscriptionEventSubject.eraseToAnyPublisher()
+    }
+
+    init() {
+        self.subscriptionEventSubject = PassthroughSubject<IncomingSubscriptionEventPublisherEvent, DataStoreError>()
+
+        let apiError = APIError.operationError(AppSyncErrorType.operationDisabled.rawValue, "", nil)
+        let dataStoreError = DataStoreError.api(apiError, nil)
+        subscriptionEventSubject.send(completion: .failure(dataStoreError))
+
+    }
+
+    func cancel() {
+    }
+}

AmplifyPlugins/DataStore/Sources/AWSDataStorePlugin/Sync/SubscriptionSync/ReconcileAndLocalSave/AWSModelReconciliationQueue.swift

@@ -209,8 +209,8 @@ final class AWSModelReconciliationQueue: ModelReconciliationQueue {
                 return
             }
             if case let .api(error, _) = dataStoreError,
-               case let APIError.operationError(_, _, underlyingError) = error,
-               isOperationDisabledError(underlyingError) {
+               case let APIError.operationError(errorMessage, _, underlyingError) = error,
+               isOperationDisabledError(errorMessage, underlyingError) {
                 log.verbose("[InitializeSubscription.3] AWSModelReconciliationQueue determined isOperationDisabledError \(modelSchema.name)")
                 modelReconciliationQueueSubject.send(.disconnected(modelName: modelSchema.name, reason: .operationDisabled))
                 return
@@ -284,7 +284,12 @@ extension AWSModelReconciliationQueue {
         return false
     }
 
-    private func isOperationDisabledError(_ error: Error?) -> Bool {
+    private func isOperationDisabledError(_ errorMessage: String?, _ error: Error?) -> Bool {
+        if let errorMessage = errorMessage,
+            case .operationDisabled = AppSyncErrorType(errorMessage) {
+            return true
+        }
+
         if let responseError = error as? GraphQLResponseError<ResponseType>,
            let graphQLError = graphqlErrors(from: responseError)?.first,
            let errorTypeValue = errorTypeValueFrom(graphQLError: graphQLError),

AmplifyPlugins/DataStore/Tests/AWSDataStorePluginTests/TestSupport/Mocks/MockAWSIncomingEventReconciliationQueue.swift

@@ -13,7 +13,7 @@ import Combine
 @testable import AWSDataStorePlugin
 
 class MockAWSIncomingEventReconciliationQueue: IncomingEventReconciliationQueue {
-    static let factory: IncomingEventReconciliationQueueFactory = { modelSchemas, api, storageAdapter, syncExpressions, auth, _, _  in
+    static let factory: IncomingEventReconciliationQueueFactory = { modelSchemas, api, storageAdapter, syncExpressions, auth, _, _, _ in
         MockAWSIncomingEventReconciliationQueue(modelSchemas: modelSchemas,
                                                 api: api,
                                                 storageAdapter: storageAdapter,

AmplifyPlugins/DataStore/Tests/DataStoreHostApp/AWSDataStorePluginIntegrationTests/Connection/DataStoreConnectionScenario1Tests.swift

@@ -42,6 +42,99 @@ class DataStoreConnectionScenario1Tests: SyncEngineIntegrationTestBase {
         let version: String = "1"
     }
 
+
+    // MARK: - Disable Real Time Updates
+
+    func testStartAndSync() async throws {
+        await setUp(withModels: TestModelRegistration(),
+                    logLevel: .verbose,
+                    dataStoreConfiguration: .custom(syncMaxRecords: 100, disableSubscriptions: { true }))
+        try await startAmplifyAndWaitForSync()
+    }
+
+    func testStartAndSyncAndRestartAndSync() async throws {
+        var disabledRealTimeUpdates = true
+        let disableSubscriptions = {
+            disabledRealTimeUpdates
+        }
+        await setUp(withModels: TestModelRegistration(),
+                    logLevel: .verbose,
+                    dataStoreConfiguration: .custom(syncMaxRecords: 100, disableSubscriptions: disableSubscriptions))
+        try await startAmplifyAndWaitForSync()
+        disabledRealTimeUpdates = false
+        try await Amplify.DataStore.stop()
+        try await Amplify.DataStore.start()
+
+        let eventReceived = expectation(description: "DataStore ready event")
+        let sink = Amplify.Hub.publisher(for: .dataStore)
+            .filter { $0.eventName == HubPayload.EventName.DataStore.ready }
+            .sink { _ in
+                eventReceived.fulfill()
+            }
+
+        try await Amplify.DataStore.start()
+
+        await fulfillment(of: [eventReceived], timeout: 10)
+    }
+
+    func testSaveReconciled() async throws {
+        await setUp(withModels: TestModelRegistration(),
+                    logLevel: .verbose,
+                    dataStoreConfiguration: .custom(syncMaxRecords: 100, disableSubscriptions: { true }))
+        try await startAmplifyAndWaitForSync()
+
+        let team = Team1(name: "name1")
+        let project = Project1(team: team)
+        let syncedTeamReceived = expectation(description: "received team from sync path")
+        var hubListener = Amplify.Hub.listen(to: .dataStore,
+                                             eventName: HubPayload.EventName.DataStore.syncReceived) { payload in
+            guard let mutationEvent = payload.data as? MutationEvent else {
+                XCTFail("Could not cast payload to mutation event")
+                return
+            }
+
+            if let syncedTeam = try? mutationEvent.decodeModel() as? Team1,
+               syncedTeam == team {
+                syncedTeamReceived.fulfill()
+            }
+        }
+        guard try await HubListenerTestUtilities.waitForListener(with: hubListener, timeout: 5.0) else {
+            XCTFail("Listener not registered for hub")
+            return
+        }
+
+        _ = try await Amplify.DataStore.save(team)
+        await fulfillment(of: [syncedTeamReceived], timeout: networkTimeout)
+
+        let syncProjectReceived = expectation(description: "received project from sync path")
+        hubListener = Amplify.Hub.listen(to: .dataStore,
+                                             eventName: HubPayload.EventName.DataStore.syncReceived) { payload in
+            guard let mutationEvent = payload.data as? MutationEvent else {
+                XCTFail("Could not cast payload to mutation event")
+                return
+            }
+
+            if let syncedProject = try? mutationEvent.decodeModel() as? Project1,
+                      syncedProject == project {
+                syncProjectReceived.fulfill()
+            }
+        }
+        guard try await HubListenerTestUtilities.waitForListener(with: hubListener, timeout: 5.0) else {
+            XCTFail("Listener not registered for hub")
+            return
+        }
+        _ = try await Amplify.DataStore.save(project)
+        await fulfillment(of: [syncProjectReceived], timeout: networkTimeout)
+
+        let queriedProjectOptional = try await Amplify.DataStore.query(Project1.self, byId: project.id)
+        guard let queriedProject = queriedProjectOptional else {
+            XCTFail("Failed")
+            return
+        }
+        XCTAssertEqual(queriedProject.id, project.id)
+        XCTAssertEqual(queriedProject.team, team)
+    }
+
     func testSaveTeamAndProjectSyncToCloud() async throws {
         await setUp(withModels: TestModelRegistration())
         try await startAmplifyAndWaitForSync()