aws-amplify · ruiguoamz · Sep 7, 2020 · Aug 25, 2020 · Aug 17, 2020 · Aug 25, 2020
@@ -353,13 +353,13 @@
 		B9FAA180238FBB5D009414B4 /* Model+Array.swift in Sources */ = {isa = PBXBuildFile; fileRef = B9FAA17F238FBB5D009414B4 /* Model+Array.swift */; };
 		B9FB05F82383740D00DE1FD4 /* DataStoreStatement.swift in Sources */ = {isa = PBXBuildFile; fileRef = B9FB05F72383740D00DE1FD4 /* DataStoreStatement.swift */; };
 		D83C5160248964780091548E /* ModelGraphQLTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = D83C515F248964780091548E /* ModelGraphQLTests.swift */; };
+		D8DD7A1D24A1CCCD001C49FD /* QuerySortInput.swift in Sources */ = {isa = PBXBuildFile; fileRef = D8DD7A1C24A1CCCD001C49FD /* QuerySortInput.swift */; };
 		FA00F68824DA37EE003E8A71 /* AuthCategoryBehavior+Combine.swift in Sources */ = {isa = PBXBuildFile; fileRef = FA00F68724DA37EE003E8A71 /* AuthCategoryBehavior+Combine.swift */; };
 		FA00F68A24DA3A43003E8A71 /* AuthCategoryDeviceBehavior+Combine.swift in Sources */ = {isa = PBXBuildFile; fileRef = FA00F68924DA3A43003E8A71 /* AuthCategoryDeviceBehavior+Combine.swift */; };
 		FA00F68C24DA3A8F003E8A71 /* AuthCategoryUserBehavior+Combine.swift in Sources */ = {isa = PBXBuildFile; fileRef = FA00F68B24DA3A8F003E8A71 /* AuthCategoryUserBehavior+Combine.swift */; };
 		FA00F68E24DA3DFF003E8A71 /* HubCategoryBehavior+Combine.swift in Sources */ = {isa = PBXBuildFile; fileRef = FA00F68D24DA3DFE003E8A71 /* HubCategoryBehavior+Combine.swift */; };
 		FA00F69024DA3F95003E8A71 /* HubCombineTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = FA00F68F24DA3F95003E8A71 /* HubCombineTests.swift */; };
 		FA00F69224DA4087003E8A71 /* PredictionsCategoryBehavior+Combine.swift in Sources */ = {isa = PBXBuildFile; fileRef = FA00F69124DA4087003E8A71 /* PredictionsCategoryBehavior+Combine.swift */; };
-		D8DD7A1D24A1CCCD001C49FD /* QuerySortInput.swift in Sources */ = {isa = PBXBuildFile; fileRef = D8DD7A1C24A1CCCD001C49FD /* QuerySortInput.swift */; };
 		FA0173352375F8A5005DDDFC /* LoggingError.swift in Sources */ = {isa = PBXBuildFile; fileRef = FA0173342375F8A5005DDDFC /* LoggingError.swift */; };
 		FA0173372375FAA5005DDDFC /* HubError.swift in Sources */ = {isa = PBXBuildFile; fileRef = FA0173362375FAA5005DDDFC /* HubError.swift */; };
 		FA05B83424CE265E0026180B /* StorageCategory+ClientBehavior+Combine.swift in Sources */ = {isa = PBXBuildFile; fileRef = FA05B83324CE265D0026180B /* StorageCategory+ClientBehavior+Combine.swift */; };

@@ -23,4 +23,44 @@ public extension HubPayload.EventName.DataStore {
     /// Dispatched when DataStore receives a sync response from the remote API via the API category. The Hub Payload
     /// will be a `MutationEvent` instance that caused the conditional save failed.
     static let conditionalSaveFailed = "DataStore.conditionalSaveFailed"
+
+    /// Dispatched when network status has changed (active or not)
+    /// The HubPayload will be a boolean value `isActive` indicating the status of network
+    // TODO: networkStatusChanged to be implemented
+    static let networkStatusChanged = "DataStore.networkStatusChanged"
+
+    /// Dispatched on DataStore start and also every time a local mutation is enqueued and processed in the outbox
+    /// HubPayload `OutboxStatusEvent` contains a boolean value `isEmpty` to notify if there are mutations in the outbox
+    static let outboxStatus = "DataStore.outboxStatus"
+
+    /// Dispatched when all of the subscriptions to syncable models have been established
+    static let subscriptionsEstablished = "DataStore.subscriptionEstablished"
+
+    /// Dispatched when DataStore is about to start sync queries
+    /// HubPayload `syncQueriesStartedEvent` contains an array of each model's `name`
+    static let syncQueriesStarted = "DataStore.syncQueriesStarted"
+
+    /// Dispatched once for each model when the model instances has been synced and updated locally
+    /// `ModelSyncedEvent` will be the HubPayload which contains `ModelName`, `isFullSync`,
+    /// `isDeltaSync` and the count for each `MutationType` (create, update, delete)
+    // TODO: modelSynced to be implemented
+    static let modelSynced = "DataStore.modelSynced"
+
+    /// Dispatched when all model instances have been synced
+    // TODO: syncQueriesReady to be implemented
+    static let syncQueriesReady = "DataStore.syncQueriesReady"
+
+    /// Dispatched when DataStore is ready, at this point all data is available
+    // TODO: ready to be implemented
+    static let ready = "DataStore.ready"
+
+    /// Dispatched when a local mutation is enqueued in the outbox
+    /// The HubPayload will be a `MutationEvent` instance about to send to remote API.
+    // TODO: outboxMutationEnqueued to be implemented
+    static let outboxMutationEnqueued = "DataStore.outboxMutationEnqueued"
+
+    /// Dispatched when a mutation in the outbox is sent to backend successfully and has been merged locally
+    /// The HubPayload will be a `MutationEvent` instance containing the newly mutated data from the remote API.
+    // TODO: outboxMutationProcessed to be implemented
+    static let outboxMutationProcessed = "DataStore.outboxMutationProcessed"
 }
@@ -0,0 +1,18 @@
+//
+// Copyright 2018-2020 Amazon.com,
+// Inc. or its affiliates. All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import Foundation
+
+/// Used as HubPayload for the `OutboxStatus`
+public struct OutboxStatusEvent {
+    /// status of outbox: empty or not
+    public let isEmpty: Bool
+
+    public init(isEmpty: Bool) {
+        self.isEmpty = isEmpty
+    }
+}
@@ -0,0 +1,18 @@
+//
+// Copyright 2018-2020 Amazon.com,
+// Inc. or its affiliates. All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import Foundation
+
+/// Used as HubPayload for the `SyncQueriesStarted`
+public struct SyncQueriesStartedEvent {
+    /// list of model names
+    public let models: [String]
+
+    public init(models: [String]) {
+        self.models = models
+    }
+}
@@ -62,19 +62,21 @@ final class AWSInitialSyncOrchestrator: InitialSyncOrchestrator {
 
         log.info("Beginning initial sync")
 
-        enqueueSyncableModels()
+        let syncableModels = ModelRegistry.models.filter { $0.schema.isSyncable }
+        enqueueSyncableModels(syncableModels)
 
         // This operation is intentionally not cancel-aware; we always want resolveCompletion to execute
         // as the last item
         syncOperationQueue.addOperation {
             self.resolveCompletion()
         }
 
+        let modelNames = syncableModels.map { $0.modelName }
+        dispatchSyncQueriesStarted(for: modelNames)
         syncOperationQueue.isSuspended = false
     }
 
-    private func enqueueSyncableModels() {
-        let syncableModels = ModelRegistry.models.filter { $0.schema.isSyncable }
+    private func enqueueSyncableModels(_ syncableModels: [Model.Type]) {
         let sortedModels = syncableModels.sortByDependencyOrder()
         for model in sortedModels {
             enqueueSyncOperation(for: model)
@@ -118,6 +120,13 @@ final class AWSInitialSyncOrchestrator: InitialSyncOrchestrator {
         completion?(.successfulVoid)
     }
 
+    private func dispatchSyncQueriesStarted(for modelNames: [String]) {
+        let syncQueriesStartedEvent = SyncQueriesStartedEvent(models: modelNames)
+        let syncQueriesStartedEventPayload = HubPayload(eventName: HubPayload.EventName.DataStore.syncQueriesStarted,
+                                                        data: syncQueriesStartedEvent)
+        Amplify.Hub.dispatch(to: .dataStore, payload: syncQueriesStartedEventPayload)
+    }
+
 }
 
 @available(iOS 13.0, *)

@@ -138,7 +138,8 @@ final class OutgoingMutationQueue: OutgoingMutationQueueBehavior {
                        mutationEventPublisher: MutationEventPublisher) {
         log.verbose(#function)
         self.api = api
-        operationQueue.isSuspended = false
+
+        checkMutationEventDatabaseStatus()
 
         // State machine notification to ".receivedSubscription" will be handled in `receive(subscription:)`
         mutationEventPublisher.publisher.subscribe(self)
@@ -187,6 +188,8 @@ final class OutgoingMutationQueue: OutgoingMutationQueueBehavior {
                     "[SyncMutationToCloudOperation] mutationEvent finished: \(mutationEvent.id); result: \(result)")
                 self.processSyncMutationToCloudResult(result, mutationEvent: mutationEvent, api: api)
         }
+
+        dispatchOutboxStatusEvent(isEmpty: false)
         operationQueue.addOperation(syncMutationToCloudOperation)
         stateMachine.notify(action: .enqueuedEvent)
     }
@@ -241,10 +244,36 @@ final class OutgoingMutationQueue: OutgoingMutationQueueBehavior {
                 self.log.verbose("mutationEvent deleted successfully")
             }
 
+            self.dispatchOutboxStatusEvent(isEmpty: true)
             self.stateMachine.notify(action: .processedEvent)
         }
     }
 
+    private func checkMutationEventDatabaseStatus() {
+        let fields = MutationEvent.keys
+        let predicate = fields.inProcess == false || fields.inProcess == nil
+
+        storageAdapter.query(MutationEvent.self,
+                             predicate: predicate,
+                             sort: nil,
+                             paginationInput: nil) { result in
+            switch result {
+            case .success(let events):
+                self.dispatchOutboxStatusEvent(isEmpty: events.isEmpty)
+            case .failure(let error):
+                log.error("Error quering mutation events: \(error)")
+            }
+            self.operationQueue.isSuspended = false
+        }
+    }
+
+    private func dispatchOutboxStatusEvent(isEmpty: Bool) {
+        let outboxStatusEvent = OutboxStatusEvent(isEmpty: isEmpty)
+        let outboxStatusEventPayload = HubPayload(eventName: HubPayload.EventName.DataStore.outboxStatus,
+                                                  data: outboxStatusEvent)
+        Amplify.Hub.dispatch(to: .dataStore, payload: outboxStatusEventPayload)
+    }
+
 }
 
 @available(iOS 13.0, *)

@@ -34,6 +34,8 @@ extension RemoteSyncEngine {
     func onReceive(receiveValue: IncomingEventReconciliationQueueEvent) {
         switch receiveValue {
         case .initialized:
+            let payload = HubPayload(eventName: HubPayload.EventName.DataStore.subscriptionsEstablished)
+            Amplify.Hub.dispatch(to: .dataStore, payload: payload)
             remoteSyncTopicPublisher.send(.subscriptionsInitialized)
             stateMachine.notify(action: .initializedSubscriptions)
         case .started:

@@ -151,6 +151,7 @@ final class AWSModelReconciliationQueue: ModelReconciliationQueue {
             modelReconciliationQueueSubject.send(.connected(modelName))
         }
     }
+
     private func receiveCompletion(_ completion: Subscribers.Completion<DataStoreError>) {
         switch completion {
         case .finished:

@@ -173,13 +173,16 @@ class DataStoreEndToEndTests: SyncEngineIntegrationTestBase {
             createdAt: date)
 
         var updatedPost = newPost
-            updatedPost.content = "UPDATED CONTENT from DataStoreEndToEndTests at \(Date())"
+        updatedPost.content = "UPDATED CONTENT from DataStoreEndToEndTests at \(Date())"
 
         let createReceived = expectation(description: "Create notification received")
         let updateLocalSuccess = expectation(description: "Update local successful")
         let conditionalReceived = expectation(description: "Conditional save failed received")
 
-        let hubListener = Amplify.Hub.listen(to: .dataStore) { payload in
+        let syncReceivedFilter = HubFilters.forEventName(HubPayload.EventName.DataStore.syncReceived)
+        let conditionalSaveFailedFilter = HubFilters.forEventName(HubPayload.EventName.DataStore.conditionalSaveFailed)
+        let filters = HubFilters.any(filters: syncReceivedFilter, conditionalSaveFailedFilter)
+        let hubListener = Amplify.Hub.listen(to: .dataStore, isIncluded: filters) { payload in
             guard let mutationEvent = payload.data as? MutationEvent
                 else {
                     XCTFail("Can't cast payload as mutation event")

@@ -0,0 +1,70 @@
+//
+// Copyright 2018-2020 Amazon.com,
+// Inc. or its affiliates. All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import XCTest
+
+import AmplifyPlugins
+import AWSPluginsCore
+
+@testable import Amplify
+@testable import AmplifyTestCommon
+@testable import AWSDataStoreCategoryPlugin
+
+@available(iOS 13.0, *)
+class DataStoreHubEventTests: HubEventsIntegrationTestBase {
+
+    /// - Given:
+    ///    - two models: Post, Comment
+    ///    - no pending MutationEvents in MutationEvent database
+    /// - When:
+    ///    - DataStore starts booting up
+    /// - Then:
+    ///    - subscriptionEstablished received, payload should be nil
+    ///    - syncQueriesStarted received, payload should be: {models: ["Post", "Comment"]}
+    ///    - outboxStatus received, payload should be {isEmpty: true}
+    func testDataStoreConfiguredDispatchesHubEvents() throws {
+
+        let subscriptionsEstablishedReceived = expectation(description: "subscriptionsEstablished received")
+        let syncQueriesStartedReceived = expectation(description: "syncQueriesStarted received")
+        let outboxStatusReceived = expectation(description: "outboxStatus received")
+
+        let hubListener = Amplify.Hub.listen(to: .dataStore) { payload in
+            if payload.eventName == HubPayload.EventName.DataStore.subscriptionsEstablished {
+                XCTAssertNil(payload.data)
+                subscriptionsEstablishedReceived.fulfill()
+            }
+
+            if payload.eventName == HubPayload.EventName.DataStore.syncQueriesStarted {
+                XCTAssertNotNil(payload.data)
+                guard let syncQueriesStartedEvent = payload.data as? SyncQueriesStartedEvent else {
+                    XCTFail("Failed to cast payload data as SyncQueriesStartedEvent")
+                    return
+                }
+                XCTAssertEqual(syncQueriesStartedEvent.models.count, 2)
+                syncQueriesStartedReceived.fulfill()
+            }
+
+            if payload.eventName == HubPayload.EventName.DataStore.outboxStatus {
+                XCTAssertNotNil(payload.data)
+                guard let outboxStatusEvent = payload.data as? OutboxStatusEvent else {
+                    XCTFail("Failed to cast payload data as OutboxStatusEvent")
+                    return
+                }
+                XCTAssertEqual(outboxStatusEvent.isEmpty, true)
+                outboxStatusReceived.fulfill()
+            }
+        }
+
+        guard try HubListenerTestUtilities.waitForListener(with: hubListener, timeout: 5.0) else {
+            XCTFail("Listener not registered for hub")
+            return
+        }
+
+        wait(for: [outboxStatusReceived, subscriptionsEstablishedReceived, syncQueriesStartedReceived],
+             timeout: networkTimeout)
+    }
+}
@@ -0,0 +1,50 @@
+//
+// Copyright 2018-2020 Amazon.com,
+// Inc. or its affiliates. All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+import XCTest
+
+import AmplifyPlugins
+import AWSMobileClient
+
+@testable import Amplify
+@testable import AmplifyTestCommon
+@testable import AWSDataStoreCategoryPlugin
+
+class HubEventsIntegrationTestBase: XCTestCase {
+
+    static let networkTimeout = TimeInterval(180)
+    let networkTimeout = HubEventsIntegrationTestBase.networkTimeout
+
+    override func setUp() {
+        super.setUp()
+
+        continueAfterFailure = false
+
+        let bundle = Bundle(for: type(of: self))
+        guard let configFile = bundle.url(forResource: "amplifyconfiguration", withExtension: "json") else {
+            XCTFail("Could not get URL for amplifyconfiguration.json from \(bundle)")
+            return
+        }
+
+        do {
+            let configData = try Data(contentsOf: configFile)
+            let amplifyConfig = try JSONDecoder().decode(AmplifyConfiguration.self, from: configData)
+            try Amplify.add(plugin: AWSAPIPlugin(modelRegistration: TestModelRegistration()))
+            try Amplify.add(plugin: AWSDataStorePlugin(modelRegistration: TestModelRegistration()))
+            try Amplify.configure(amplifyConfig)
+        } catch {
+            XCTFail(String(describing: error))
+            return
+        }
+    }
+
+    override func tearDown() {
+        sleep(1)
+        print("Amplify reset")
+        Amplify.reset()
+    }
+}