feat(datastore): Dispatch networkStatus event

Amplify/Categories/DataStore/DataStoreCategory+HubPayloadEventName.swift

@@ -38,6 +38,10 @@ public extension HubPayload.EventName.DataStore {
     /// HubPayload `syncQueriesStartedEvent` contains an array of each model's `name`
     static let syncQueriesStarted = "DataStore.syncQueriesStarted"
 
+    /// Dispatched when DataStore starts and everytime network status changes
+    /// HubPayload `NetworkStatusEvent` contains a boolean value `active` to notify network status
+    static let networkStatus = "DataStore.networkStatus"
+
     /// Dispatched when a local mutation is enqueued into the outgoing mutation queue `outbox`
     /// HubPayload `outboxMutationEvent` contains the name and instance of the model
     static let outboxMutationEnqueued = "DataStore.outboxMutationEnqueued"

AmplifyPlugins/API/AWSAPICategoryPlugin/Reachability/NetworkReachabilityNotifier.swift

@@ -15,7 +15,7 @@ class NetworkReachabilityNotifier {
     private var reachability: NetworkReachabilityProviding?
     private var allowsCellularAccess = true
 
-    let reachabilityPublisher = PassthroughSubject<ReachabilityUpdate, Never>()
+    let reachabilityPublisher = CurrentValueSubject<ReachabilityUpdate, Never>(ReachabilityUpdate(isOnline: false))
     var publisher: AnyPublisher<ReachabilityUpdate, Never> {
         return reachabilityPublisher.eraseToAnyPublisher()
     }

AmplifyPlugins/API/AWSAPICategoryPluginTests/Reachability/NetworkReachabilityNotifierTests.swift

@@ -31,45 +31,60 @@ class NetworkReachabilityNotifierTests: XCTestCase {
 
     func testWifiConnectivity() {
         MockReachability.iConnection = .wifi
-        let expect = expectation(description: ".sink receives value")
+        let expect = expectation(description: ".sink receives values")
+        var values = [Bool]()
         let cancellable = notifier.publisher.sink(receiveCompletion: { _ in
             XCTFail("Not expecting any error")
         }, receiveValue: { value in
-            XCTAssert(value.isOnline)
-            expect.fulfill()
+            values.append(value.isOnline)
+            if values.count == 2 {
+                XCTAssertFalse(values[0])
+                XCTAssertTrue(values[1])
+                expect.fulfill()
+            }
         })
         notification = Notification.init(name: .reachabilityChanged)
         NotificationCenter.default.post(notification)
 
         waitForExpectations(timeout: 1.0)
         cancellable.cancel()
     }
+
     func testCellularConnectivity() {
         MockReachability.iConnection = .wifi
-        let expect = expectation(description: ".sink receives value")
+        let expect = expectation(description: ".sink receives values")
+        var values = [Bool]()
         let cancellable = notifier.publisher.sink(receiveCompletion: { _ in
             XCTFail("Not expecting any error")
         }, receiveValue: { value in
-            XCTAssert(value.isOnline)
-            expect.fulfill()
+            values.append(value.isOnline)
+            if values.count == 2 {
+                XCTAssertFalse(values[0])
+                XCTAssertTrue(values[1])
+                expect.fulfill()
+            }
         })
 
         notification = Notification.init(name: .reachabilityChanged)
         NotificationCenter.default.post(notification)
 
         waitForExpectations(timeout: 1.0)
         cancellable.cancel()
-
     }
 
     func testNoConnectivity() {
         MockReachability.iConnection = .unavailable
-        let expect = expectation(description: ".sink receives value")
+        let expect = expectation(description: ".sink receives values")
+        var values = [Bool]()
         let cancellable = notifier.publisher.sink(receiveCompletion: { _ in
             XCTFail("Not expecting any error")
         }, receiveValue: { value in
-            XCTAssertFalse(value.isOnline)
-            expect.fulfill()
+            values.append(value.isOnline)
+            if values.count == 2 {
+                XCTAssertFalse(values[0])
+                XCTAssertFalse(values[1])
+                expect.fulfill()
+            }
         })
 
         notification = Notification.init(name: .reachabilityChanged)
@@ -81,18 +96,21 @@ class NetworkReachabilityNotifierTests: XCTestCase {
 
     func testWifiConnectivity_publisherGoesOutOfScope() {
         MockReachability.iConnection = .wifi
-        let expect = expectation(description: ".sink receives value")
+        let defaultValueExpect = expectation(description: ".sink receives default value")
+        let completeExpect = expectation(description: ".sink receives completion")
         let cancellable = notifier.publisher.sink(receiveCompletion: { _ in
-            expect.fulfill()
-        }, receiveValue: { _ in
-            XCTAssertFalse(true)
+            completeExpect.fulfill()
+        }, receiveValue: { value in
+            XCTAssertFalse(value.isOnline)
+            defaultValueExpect.fulfill()
         })
 
+        wait(for: [defaultValueExpect], timeout: 1.0)
         notifier = nil
         notification = Notification.init(name: .reachabilityChanged)
         NotificationCenter.default.post(notification)
 
-        waitForExpectations(timeout: 1.0)
+        wait(for: [completeExpect], timeout: 1.0)
         cancellable.cancel()
     }
 }

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/Events/NetworkStatusEvent.swift

@@ -0,0 +1,16 @@
+//
+// Copyright 2018-2020 Amazon.com,
+// Inc. or its affiliates. All Rights Reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+/// Used as HubPayload for the `NetworkStatus`
+public struct NetworkStatusEvent {
+    /// status of network: true if network is active
+    public let active: Bool
+
+    public init(active: Bool) {
+        self.active = active
+    }
+}

AmplifyPlugins/DataStore/AWSDataStoreCategoryPlugin/Sync/RemoteSyncEngine.swift

@@ -48,6 +48,7 @@ class RemoteSyncEngine: RemoteSyncEngineBehavior {
     private var stateMachineSink: AnyCancellable?
 
     var networkReachabilityPublisher: AnyPublisher<ReachabilityUpdate, Never>?
+    private var networkReachabilitySink: AnyCancellable?
     var mutationRetryNotifier: MutationRetryNotifier?
     let requestRetryablePolicy: RequestRetryablePolicy
     var currentAttemptNumber: Int
@@ -184,6 +185,17 @@ class RemoteSyncEngine: RemoteSyncEngineBehavior {
         self.api = api
         self.auth = auth
 
+        if networkReachabilityPublisher == nil,
+            let reachability = api as? APICategoryReachabilityBehavior {
+            do {
+                networkReachabilityPublisher = try reachability.reachabilityPublisher()
+                networkReachabilitySink = networkReachabilityPublisher?
+                    .sink(receiveValue: onReceiveNetworkStatus(networkStatus:))
+            } catch {
+                log.error("\(#function): Unable to listen on reachability: \(error)")
+            }
+        }
+
         remoteSyncTopicPublisher.send(.storageAdapterAvailable)
         stateMachine.notify(action: .receivedStart)
     }
@@ -327,6 +339,13 @@ class RemoteSyncEngine: RemoteSyncEngineBehavior {
         stateMachine.notify(action: .notifiedSyncStarted)
     }
 
+    private func onReceiveNetworkStatus(networkStatus: ReachabilityUpdate) {
+        let networkStatusEvent = NetworkStatusEvent(active: networkStatus.isOnline)
+        let networkStatusEventPayload = HubPayload(eventName: HubPayload.EventName.DataStore.networkStatus,
+                                                   data: networkStatusEvent)
+        Amplify.Hub.dispatch(to: .dataStore, payload: networkStatusEventPayload)
+    }
+
     func reset(onComplete: () -> Void) {
         let group = DispatchGroup()
 

AmplifyPlugins/DataStore/AWSDataStoreCategoryPluginIntegrationTests/DataStoreHubEventsTests.swift

@@ -23,16 +23,27 @@ class DataStoreHubEventTests: HubEventsIntegrationTestBase {
     /// - When:
     ///    - DataStore's remote sync engine is initialized
     /// - Then:
+    ///    - networkStatus received, payload should be: {active: true}
     ///    - 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 networkStatusReceived = expectation(description: "networkStatus received")
         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
+        let hubListener = Amplify.Hub.publisher(for: .dataStore).sink { payload in
+            if payload.eventName == HubPayload.EventName.DataStore.networkStatus {
+                guard let networkStatusEvent = payload.data as? NetworkStatusEvent else {
+                    XCTFail("Failed to cast payload data as NetworkStatusEvent")
+                    return
+                }
+                XCTAssertEqual(networkStatusEvent.active, true)
+                networkStatusReceived.fulfill()
+            }
+
             if payload.eventName == HubPayload.EventName.DataStore.subscriptionsEstablished {
                 XCTAssertNil(payload.data)
                 subscriptionsEstablishedReceived.fulfill()
@@ -57,13 +68,10 @@ class DataStoreHubEventTests: HubEventsIntegrationTestBase {
             }
         }
 
-        guard try HubListenerTestUtilities.waitForListener(with: hubListener, timeout: 5.0) else {
-            XCTFail("Listener not registered for hub")
-            return
-        }
+        startAmplify()
 
         waitForExpectations(timeout: networkTimeout, handler: nil)
-        Amplify.Hub.removeListener(hubListener)
+        hubListener.cancel()
     }
 
     /// - Given:
@@ -83,7 +91,7 @@ class DataStoreHubEventTests: HubEventsIntegrationTestBase {
         let outboxMutationEnqueuedReceived = expectation(description: "outboxMutationEnqueued received")
         let outboxMutationProcessedReceived = expectation(description: "outboxMutationProcessed received")
 
-        let hubListener = Amplify.Hub.listen(to: .dataStore) { payload in
+        let hubListener = Amplify.Hub.publisher(for: .dataStore).sink { payload in
             if payload.eventName == HubPayload.EventName.DataStore.outboxMutationEnqueued {
                 guard let outboxMutationEnqueuedEvent = payload.data as? OutboxMutationEvent else {
                     XCTFail("Failed to cast payload data as OutboxMutationEvent")
@@ -110,15 +118,11 @@ class DataStoreHubEventTests: HubEventsIntegrationTestBase {
                 outboxMutationProcessedReceived.fulfill()
             }
         }
-
-        guard try HubListenerTestUtilities.waitForListener(with: hubListener, timeout: 5.0) else {
-            XCTFail("Listener not registered for hub")
-            return
-        }
-
+
+        startAmplify()
         Amplify.DataStore.save(post) { _ in }
 
         waitForExpectations(timeout: networkTimeout, handler: nil)
-        Amplify.Hub.removeListener(hubListener)
+        hubListener.cancel()
     }
 }

AmplifyPlugins/DataStore/AWSDataStoreCategoryPluginIntegrationTests/TestSupport/HubEventsIntegrationTestBase.swift

@@ -21,9 +21,11 @@ class HubEventsIntegrationTestBase: XCTestCase {
 
     override func setUp() {
         super.setUp()
-
         continueAfterFailure = false
+        Amplify.reset()
+    }
 
+    func startAmplify() {
         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)")
@@ -33,8 +35,8 @@ class HubEventsIntegrationTestBase: XCTestCase {
         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.add(plugin: AWSAPIPlugin(modelRegistration: TestModelRegistration()))
             try Amplify.configure(amplifyConfig)
         } catch {
             XCTFail(String(describing: error))

AmplifyPlugins/DataStore/AWSDataStoreCategoryPluginTests/Sync/RemoteSyncEngineTests.swift

@@ -25,6 +25,7 @@ class RemoteSyncEngineTests: XCTestCase {
 
     override func setUp() {
         super.setUp()
+        apiPlugin = MockAPICategoryPlugin()
         MockAWSInitialSyncOrchestrator.reset()
         storageAdapter = MockSQLiteStorageEngineAdapter()
         let mockOutgoingMutationQueue = MockOutgoingMutationQueue()
@@ -118,7 +119,7 @@ class RemoteSyncEngineTests: XCTestCase {
         MockAWSInitialSyncOrchestrator.setResponseOnSync(result:
             .failure(DataStoreError.internalOperation("forceError", "none", nil)))
 
-        remoteSyncEngine.start()
+        remoteSyncEngine.start(api: apiPlugin)
 
         wait(for: [storageAdapterAvailable,
                    subscriptionsPaused,
@@ -177,7 +178,7 @@ class RemoteSyncEngineTests: XCTestCase {
                 }
             })
 
-        remoteSyncEngine.start()
+        remoteSyncEngine.start(api: apiPlugin)
 
         wait(for: [storageAdapterAvailable,
                    subscriptionsPaused,
@@ -248,7 +249,7 @@ class RemoteSyncEngineTests: XCTestCase {
                 }
             })
 
-        remoteSyncEngine.start()
+        remoteSyncEngine.start(api: apiPlugin)
 
         wait(for: [storageAdapterAvailable,
                    subscriptionsPaused,
@@ -325,7 +326,7 @@ class RemoteSyncEngineTests: XCTestCase {
                 }
             })
 
-        remoteSyncEngine.start()
+        remoteSyncEngine.start(api: apiPlugin)
 
         wait(for: [storageAdapterAvailable,
                    subscriptionsPaused,

AmplifyPlugins/DataStore/DataStoreCategoryPlugin.xcodeproj/project.pbxproj

@@ -106,6 +106,7 @@
 		D8D900A4249DB599004042E7 /* QuerySort+SQLite.swift in Sources */ = {isa = PBXBuildFile; fileRef = D8D900A3249DB599004042E7 /* QuerySort+SQLite.swift */; };
 		D8E9992425013C2F0006170A /* DataStoreHubEventsTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = D8E9992325013C2F0006170A /* DataStoreHubEventsTests.swift */; };
 		D8E99926250142790006170A /* HubEventsIntegrationTestBase.swift in Sources */ = {isa = PBXBuildFile; fileRef = D8E99925250142790006170A /* HubEventsIntegrationTestBase.swift */; };
+		D8FF6207250EB4C3000BFB4B /* NetworkStatusEvent.swift in Sources */ = {isa = PBXBuildFile; fileRef = D8FF6206250EB4C3000BFB4B /* NetworkStatusEvent.swift */; };
 		FA0427C82396C27400D25AB0 /* SyncEngineStartupTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = FA0427C72396C27400D25AB0 /* SyncEngineStartupTests.swift */; };
 		FA0427CA2396C35500D25AB0 /* InitialSyncOrchestrator.swift in Sources */ = {isa = PBXBuildFile; fileRef = FA0427C92396C35500D25AB0 /* InitialSyncOrchestrator.swift */; };
 		FA0427CC2396C7E400D25AB0 /* InitialSyncOrchestratorTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = FA0427CB2396C7E400D25AB0 /* InitialSyncOrchestratorTests.swift */; };
@@ -325,6 +326,7 @@
 		D8D900A3249DB599004042E7 /* QuerySort+SQLite.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "QuerySort+SQLite.swift"; sourceTree = "<group>"; };
 		D8E9992325013C2F0006170A /* DataStoreHubEventsTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = DataStoreHubEventsTests.swift; sourceTree = "<group>"; };
 		D8E99925250142790006170A /* HubEventsIntegrationTestBase.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = HubEventsIntegrationTestBase.swift; sourceTree = "<group>"; };
+		D8FF6206250EB4C3000BFB4B /* NetworkStatusEvent.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = NetworkStatusEvent.swift; sourceTree = "<group>"; };
 		DCC3AA75D77D0DB916EC42DB /* Pods-HostApp.release.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-HostApp.release.xcconfig"; path = "Target Support Files/Pods-HostApp/Pods-HostApp.release.xcconfig"; sourceTree = "<group>"; };
 		EA320D973669D3843FDF755E /* Pods_HostApp_AWSDataStoreCategoryPluginAuthIntegrationTests.framework */ = {isa = PBXFileReference; explicitFileType = wrapper.framework; includeInIndex = 0; path = Pods_HostApp_AWSDataStoreCategoryPluginAuthIntegrationTests.framework; sourceTree = BUILT_PRODUCTS_DIR; };
 		EA53DF78CC578B7C81E72D83 /* Pods-AWSDataStoreCategoryPlugin-AWSDataStoreCategoryPluginTests.debug.xcconfig */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = text.xcconfig; name = "Pods-AWSDataStoreCategoryPlugin-AWSDataStoreCategoryPluginTests.debug.xcconfig"; path = "Target Support Files/Pods-AWSDataStoreCategoryPlugin-AWSDataStoreCategoryPluginTests/Pods-AWSDataStoreCategoryPlugin-AWSDataStoreCategoryPluginTests.debug.xcconfig"; sourceTree = "<group>"; };
@@ -666,6 +668,7 @@
 			isa = PBXGroup;
 			children = (
 				D838AB4524FF268400BF4940 /* SyncQueriesStartedEvent.swift */,
+				D8FF6206250EB4C3000BFB4B /* NetworkStatusEvent.swift */,
 				D838AB4324FF264600BF4940 /* OutboxStatusEvent.swift */,
 				D88666A325070FC6000F7A14 /* OutboxMutationEvent.swift */,
 			);
@@ -1516,6 +1519,7 @@
 				FA5D4CEF238AFCBC00D2F54A /* AWSDataStorePlugin+DataStoreSubscribeBehavior.swift in Sources */,
 				FAAFAF3923905F35002CF932 /* MutationEventPublisher.swift in Sources */,
 				6BB93F07244BA44B00ED1FC3 /* MutationEventClearState.swift in Sources */,
+				D8FF6207250EB4C3000BFB4B /* NetworkStatusEvent.swift in Sources */,
 				2149E5C82388684F00873955 /* SQLStatement.swift in Sources */,
 				FAED5742238B52CE008EBED8 /* ModelStorageBehavior.swift in Sources */,
 				2149E5C62388684F00873955 /* StorageEngineAdapter.swift in Sources */,

AmplifyTestCommon/Mocks/MockAPICategoryPlugin.swift

@@ -9,7 +9,7 @@ import Amplify
 import Combine
 import Foundation
 
-class MockAPICategoryPlugin: MessageReporter, APICategoryPlugin {
+class MockAPICategoryPlugin: MessageReporter, APICategoryPlugin, APICategoryReachabilityBehavior {
     var responders = [ResponderKeys: Any]()
 
     // MARK: - Properties
@@ -110,7 +110,7 @@ class MockAPICategoryPlugin: MessageReporter, APICategoryPlugin {
 
     @available(iOS 13.0, *)
     public func reachabilityPublisher() -> AnyPublisher<ReachabilityUpdate, Never>? {
-        return nil
+        return Just(ReachabilityUpdate(isOnline: true)).eraseToAnyPublisher()
     }
 
     // MARK: - REST methods