@fix(data): Fix reachability debouncer causing delay on local inserts (#2464)

Author: tylerjroach

aws-datastore/src/androidTest/java/com/amplifyframework/datastore/BasicCloudSyncInstrumentationTest.java

@@ -216,13 +216,16 @@ public void syncDownFromCloudIsWorking() throws AmplifyException {
     }
 
     /**
-     * Verify that updating an item shortly after creating it succeeds.  This can be tricky because the _version
-     * returned in the response from the create request must be included in the input for the subsequent update request.
+     * Verify that updating an item shortly after creating it succeeds locally.
+     *
+     * Note: If this test periodically fails, consider the immediate update save may still be in process since
+     * we are only waiting on 1 hub event. Because we call back to back, I haven't seen this happen yet.
+     *
      * @throws DataStoreException On failure to save or query items from DataStore.
      * @throws ApiException On failure to query the API.
      */
     @Test
-    public void updateAfterCreate() throws DataStoreException, ApiException {
+    public void createThenUpdate() throws DataStoreException, ApiException {
         // Setup
         BlogOwner richard = BlogOwner.builder()
                 .name("Richard")
@@ -232,16 +235,16 @@ public void updateAfterCreate() throws DataStoreException, ApiException {
                 .build();
         String modelName = BlogOwner.class.getSimpleName();
 
-        // Expect two mutations to be published to AppSync.
+        // Expect at least 1 mutation to be published to AppSync.
         HubAccumulator richardAccumulator =
-            HubAccumulator.create(HubChannel.DATASTORE, publicationOf(modelName, richard.getId()), 2)
+            HubAccumulator.create(HubChannel.DATASTORE, publicationOf(modelName, richard.getId()), 1)
                 .start();
 
         // Create an item, then update it and save it again.
         dataStore.save(richard);
         dataStore.save(updatedRichard);
 
-        // Verify that 2 mutations were published.
+        // Verify that at least 1 mutations was published.
         richardAccumulator.await(60, TimeUnit.SECONDS);
 
         // Verify that the updatedRichard is saved in the DataStore.
@@ -254,7 +257,56 @@ public void updateAfterCreate() throws DataStoreException, ApiException {
     }
 
     /**
-     * Verify that updating a different field of an item shortly after creating it succeeds.
+     * Verify that updating an item shortly after creating it succeeds. This can be tricky because the _version
+     * returned in the response from the create request must be included in the input for the subsequent update request.
+     * @throws DataStoreException On failure to save or query items from DataStore.
+     * @throws ApiException On failure to query the API.
+     */
+    @Test
+    public void createWaitThenUpdate() throws DataStoreException, ApiException {
+        // Setup
+        BlogOwner richard = BlogOwner.builder()
+                                .name("Richard")
+                                .build();
+        BlogOwner updatedRichard = richard.copyOfBuilder()
+                                       .name("Richard McClellan")
+                                       .build();
+        String modelName = BlogOwner.class.getSimpleName();
+
+        HubAccumulator accumulator1 =
+            HubAccumulator.create(HubChannel.DATASTORE, publicationOf(modelName, richard.getId()), 1);
+        HubAccumulator accumulator2 =
+            HubAccumulator.create(HubChannel.DATASTORE, publicationOf(modelName, richard.getId()), 1);
+
+        // Create an item, then update it and save it again.
+        accumulator1.start();
+        dataStore.save(richard);
+
+        // Verify first save published
+        accumulator1.await(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+
+        // Update item and save
+        accumulator2.start();
+        dataStore.save(updatedRichard);
+
+        // Verify update published
+        accumulator2.await(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+
+        // Verify that the updatedRichard is saved in the DataStore.
+        BlogOwner localRichard = dataStore.get(BlogOwner.class, richard.getId());
+        ModelAssert.assertEqualsIgnoringTimestamps(updatedRichard, localRichard);
+
+        // Verify that the updatedRichard is saved on the backend.
+        BlogOwner remoteRichard = api.get(BlogOwner.class, richard.getId());
+        ModelAssert.assertEqualsIgnoringTimestamps(updatedRichard, remoteRichard);
+    }
+
+    /**
+     * Verify that updating a different field of an item immediately after creating it succeeds.
+     *
+     * Note: If this test periodically fails, consider the immediate update save may still be in process since
+     * we are only waiting on 1 hub event. Because we call back to back, I haven't seen this happen yet.
+     *
      * @throws DataStoreException On failure to save or query items from DataStore.
      * @throws ApiException On failure to query the API.
      */
@@ -269,17 +321,61 @@ public void createThenUpdateDifferentField() throws DataStoreException, ApiExcep
                 .build();
         String modelName = BlogOwner.class.getSimpleName();
 
-        // Expect two mutations to be published to AppSync.
+        // Expect at least 1 mutation to be published to AppSync.
         HubAccumulator accumulator =
-                HubAccumulator.create(HubChannel.DATASTORE, publicationOf(modelName, owner.getId()), 2)
+                HubAccumulator.create(HubChannel.DATASTORE, publicationOf(modelName, owner.getId()), 1)
                         .start();
 
         // Create an item, then update it with different field and save it again.
         dataStore.save(owner);
         dataStore.save(updatedOwner);
 
-        // Verify that 2 mutations were published.
-        accumulator.await(60, TimeUnit.SECONDS);
+        // Verify that mutation(s) were published.
+        accumulator.await(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+
+        // Verify that the updatedOwner is saved in the DataStore.
+        BlogOwner localOwner = dataStore.get(BlogOwner.class, owner.getId());
+        ModelAssert.assertEqualsIgnoringTimestamps(updatedOwner, localOwner);
+
+        // Verify that the updatedOwner is saved on the backend.
+        BlogOwner remoteOwner = api.get(BlogOwner.class, owner.getId());
+        ModelAssert.assertEqualsIgnoringTimestamps(updatedOwner, remoteOwner);
+    }
+
+    /**
+     * Verify that updating a different field of an item succeeds, after verifying the initial item has been published.
+     * @throws DataStoreException On failure to save or query items from DataStore.
+     * @throws ApiException On failure to query the API.
+     */
+    @Test
+    public void createWaitThenUpdateDifferentField() throws DataStoreException, ApiException {
+        // Setup
+        BlogOwner owner = BlogOwner.builder()
+                              .name("Richard")
+                              .build();
+        BlogOwner updatedOwner = owner.copyOfBuilder()
+                                     .wea("pon")
+                                     .build();
+        String modelName = BlogOwner.class.getSimpleName();
+
+        HubAccumulator accumulator1 =
+            HubAccumulator.create(HubChannel.DATASTORE, publicationOf(modelName, owner.getId()), 1);
+        HubAccumulator accumulator2 =
+            HubAccumulator.create(HubChannel.DATASTORE, publicationOf(modelName, owner.getId()), 1);
+
+        // Create an item, then
+        accumulator1.start();
+        dataStore.save(owner);
+
+        // Verify save published
+        accumulator1.await(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+
+        // Update item with different field and save it again.
+        accumulator2.start();
+        dataStore.save(updatedOwner);
+
+        // Verify update save published
+        accumulator2.await(TIMEOUT_SECONDS, TimeUnit.SECONDS);
 
         // Verify that the updatedOwner is saved in the DataStore.
         BlogOwner localOwner = dataStore.get(BlogOwner.class, owner.getId());
@@ -333,31 +429,52 @@ public void create1ThenCreate2ThenUpdate2() throws DataStoreException, ApiExcept
     }
 
     /**
-     * Verify that creating a new item, then immediately delete succeeds.
+     * Verify that creating a new item, then immediately deleting succeeds.
      * @throws DataStoreException On failure to save or query items from DataStore.
      * @throws ApiException On failure to query the API.
      */
     @Test
     public void createThenDelete() throws DataStoreException, ApiException {
+        // Setup
+        BlogOwner owner = BlogOwner.builder()
+                .name("Jean")
+                .build();
+
+        dataStore.save(owner);
+        dataStore.delete(owner);
+
+        // Verify that the owner is deleted from the local data store.
+        assertThrows(NoSuchElementException.class, () -> dataStore.get(BlogOwner.class, owner.getId()));
+    }
+
+    /**
+     * Verify that creating a new item, waiting for it to post, then immediately delete succeeds.
+     * @throws DataStoreException On failure to save or query items from DataStore.
+     * @throws ApiException On failure to query the API.
+     */
+    @Test
+    public void createWaitThenDelete() throws DataStoreException, ApiException {
         // Setup
         BlogOwner owner = BlogOwner.builder()
                 .name("Jean")
                 .build();
         String modelName = BlogOwner.class.getSimpleName();
-        
-        HubAccumulator accumulator =
-                HubAccumulator.create(HubChannel.DATASTORE, publicationOf(modelName, owner.getId()), 2)
-                        .start();
+
+        HubAccumulator accumulator1 =
+                HubAccumulator.create(HubChannel.DATASTORE, publicationOf(modelName, owner.getId()), 1);
+        HubAccumulator accumulator2 =
+            HubAccumulator.create(HubChannel.DATASTORE, publicationOf(modelName, owner.getId()), 1);
+
+        accumulator1.start();
         dataStore.save(owner);
+        accumulator1.awaitFirst(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+
+        accumulator2.start();
         dataStore.delete(owner);
-        
-        accumulator.await(TIMEOUT_SECONDS, TimeUnit.SECONDS);
-        
+        accumulator2.await(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+
         // Verify that the owner is deleted from the local data store.
         assertThrows(NoSuchElementException.class, () -> dataStore.get(BlogOwner.class, owner.getId()));
-        
-        // Note: Currently, default GraphQL resolvers do not filter records that have been deleted.
-        // Therefore, calling api to get the item at this point would still succeed.
     }
 
     /**
@@ -457,6 +574,10 @@ public void createWaitThenUpdate10TimesWithPredicate() throws DataStoreException
     /**
      * Create new item, then immediately update a different field. 
      * Wait for sync round trip. Then update the first field.
+     *
+     * Note: If this test periodically fails, consider the immediate update save may still be in process since
+     * we are only waiting on 1 hub event. Because we call back to back, I haven't seen this happen yet.
+     *
      * @throws DataStoreException On failure to save or query items from DataStore.
      * @throws ApiException On failure to query the API.
      */
@@ -466,9 +587,10 @@ public void createItemThenUpdateThenWaitThenUpdate() throws DataStoreException,
         BlogOwner owner = BlogOwner.builder().name("ownerName").build();
         BlogOwner updatedOwner = owner.copyOfBuilder().wea("pon").build();
         String modelName = BlogOwner.class.getSimpleName();
-        
+
+        // Expect at least 1 update (2 is possible)
         HubAccumulator accumulator =
-                HubAccumulator.create(HubChannel.DATASTORE, publicationOf(modelName, owner.getId()), 2)
+                HubAccumulator.create(HubChannel.DATASTORE, publicationOf(modelName, owner.getId()), 1)
                         .start();
         // Create new and then immediately update
         dataStore.save(owner);

aws-datastore/src/main/java/com/amplifyframework/datastore/syncengine/ReachabilityMonitor.kt

@@ -62,32 +62,34 @@ private class ReachabilityMonitorImpl constructor(val schedulerProvider: Schedul
 
     override fun configure(context: Context, connectivityProvider: ConnectivityProvider) {
         this.connectivityProvider = connectivityProvider
-        connectivityProvider.registerDefaultNetworkCallback(
-            context,
-            object : NetworkCallback() {
-                override fun onAvailable(network: Network) {
-                    subject.onNext(true)
-                }
+        val observable = Observable.create { emitter ->
+            connectivityProvider.registerDefaultNetworkCallback(
+                context,
+                object : NetworkCallback() {
+                    override fun onAvailable(network: Network) {
+                        emitter.onNext(true)
+                    }
 
-                override fun onLost(network: Network) {
-                    subject.onNext(false)
+                    override fun onLost(network: Network) {
+                        emitter.onNext(false)
+                    }
                 }
-            }
-        )
+            )
+            emitter.onNext(connectivityProvider.hasActiveNetwork)
+        }
+        observable.debounce(250, TimeUnit.MILLISECONDS, schedulerProvider.computation())
+            .distinctUntilChanged()
+            .subscribe(subject)
     }
 
     override fun getObservable(): Observable<Boolean> {
-        connectivityProvider?.let { connectivityProvider ->
-            return subject.subscribeOn(schedulerProvider.io())
-                .doOnSubscribe { subject.onNext(connectivityProvider.hasActiveNetwork) }
-                .debounce(250, TimeUnit.MILLISECONDS, schedulerProvider.computation())
-                .distinctUntilChanged()
-        } ?: run {
+        if (connectivityProvider == null) {
             throw DataStoreException(
                 "ReachabilityMonitor has not been configured.",
                 "Call ReachabilityMonitor.configure() before calling ReachabilityMonitor.getObservable()"
             )
         }
+        return subject.subscribeOn(schedulerProvider.io())
     }
 }