Fix RxLint errors by adding onError handlers and disposing of subscriptions

Author: mattcreaser

aws-datastore/src/main/java/com/amplifyframework/datastore/AWSDataStorePlugin.java

@@ -15,6 +15,7 @@
 
 package com.amplifyframework.datastore;
 
+import android.annotation.SuppressLint;
 import android.content.Context;
 import androidx.annotation.NonNull;
 import androidx.annotation.Nullable;
@@ -65,11 +66,14 @@
 import java.util.concurrent.TimeUnit;
 
 import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.disposables.CompositeDisposable;
+import io.reactivex.rxjava3.disposables.Disposable;
 import io.reactivex.rxjava3.schedulers.Schedulers;
 
 /**
  * An AWS implementation of the {@link DataStorePlugin}.
  */
+@SuppressLint("RxLeakedSubscription")
 public final class AWSDataStorePlugin extends DataStorePlugin<Void> {
     private static final Logger LOG = Amplify.Logging.logger(CategoryType.DATASTORE, "amplify:aws-datastore");
     private static final long LIFECYCLE_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(5);
@@ -98,6 +102,12 @@ public final class AWSDataStorePlugin extends DataStorePlugin<Void> {
 
     private final ReachabilityMonitor reachabilityMonitor;
 
+    // Subscriptions that should be disposed when datastore is stopped
+    private final CompositeDisposable startedDisposables = new CompositeDisposable();
+
+    // Subscriptions that have the same lifetime as the plugin
+    private final CompositeDisposable pluginDisposables = new CompositeDisposable();
+
     private AWSDataStorePlugin(
             @NonNull ModelProvider modelProvider,
             @NonNull SchemaRegistry schemaRegistry,
@@ -286,7 +296,11 @@ private void configure(Context context, DataStoreConfiguration configuration) {
 
         reachabilityMonitor.configure(context);
 
-        waitForInitialization().subscribe(this::observeNetworkStatus);
+        Disposable subscription = waitForInitialization().subscribe(
+            this::observeNetworkStatus,
+            error -> LOG.error("Datastore did not initialize", error)
+        );
+        pluginDisposables.add(subscription);
     }
 
     private void publishNetworkStatusEvent(boolean active) {
@@ -295,16 +309,27 @@ private void publishNetworkStatusEvent(boolean active) {
     }
 
     private void observeNetworkStatus() {
-        reachabilityMonitor.getObservable()
-                .subscribe(this::publishNetworkStatusEvent);
+        Disposable subscription = reachabilityMonitor.getObservable()
+                .subscribe(
+                    this::publishNetworkStatusEvent,
+                    error -> LOG.warn("Unable to subscribe to network status events", error)
+                );
+        pluginDisposables.add(subscription);
     }
 
+    @SuppressLint("CheckResult")
     @WorkerThread
     @Override
     public void initialize(@NonNull Context context) throws AmplifyException {
         try {
-            initializeStorageAdapter(context)
-                .blockingAwait(LIFECYCLE_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+            boolean initialized = initializeStorageAdapter(context)
+                                      .blockingAwait(LIFECYCLE_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+            if (!initialized) {
+                throw new DataStoreException(
+                    "Storage adapter did not initialize within allotted timeout",
+                    AmplifyException.REPORT_BUG_TO_AWS_SUGGESTION
+                );
+            }
         } catch (Throwable initError) {
             throw new AmplifyException(
                 "Failed to initialize the local storage adapter for the DataStore plugin.",
@@ -327,8 +352,8 @@ private Completable initializeStorageAdapter(Context context) {
     }
 
     private Completable waitForInitialization() {
-        return Completable.fromAction(() -> categoryInitializationsPending.await())
-            .timeout(LIFECYCLE_TIMEOUT_MS, TimeUnit.MILLISECONDS)
+        return Completable.fromAction(categoryInitializationsPending::await)
+            .timeout(LIFECYCLE_TIMEOUT_MS, TimeUnit.MILLISECONDS, Schedulers.io())
             .subscribeOn(Schedulers.io())
             .doOnComplete(() -> LOG.info("DataStore plugin initialized."))
             .doOnError(error -> LOG.error("DataStore initialization timed out.", error));
@@ -339,27 +364,30 @@ private Completable waitForInitialization() {
      */
     @Override
     public void start(@NonNull Action onComplete, @NonNull Consumer<DataStoreException> onError) {
-        waitForInitialization()
+        Disposable subscription = waitForInitialization()
             .andThen(orchestrator.start())
             .subscribeOn(Schedulers.io())
             .subscribe(
                 onComplete::call,
                 error -> onError.accept(new DataStoreException("Failed to start DataStore.", error, "Retry."))
             );
+        startedDisposables.add(subscription);
     }
 
     /**
      * {@inheritDoc}
      */
     @Override
     public void stop(@NonNull Action onComplete, @NonNull Consumer<DataStoreException> onError) {
-        waitForInitialization()
+        startedDisposables.dispose();
+        Disposable subscription = waitForInitialization()
             .andThen(orchestrator.stop())
             .subscribeOn(Schedulers.io())
             .subscribe(
                 onComplete::call,
                 error -> onError.accept(new DataStoreException("Failed to stop DataStore.", error, "Retry."))
             );
+        startedDisposables.add(subscription);
     }
 
     /**
@@ -372,12 +400,19 @@ public void stop(@NonNull Action onComplete, @NonNull Consumer<DataStoreExceptio
      */
     @Override
     public void clear(@NonNull Action onComplete, @NonNull Consumer<DataStoreException> onError) {
-        stop(() -> Completable.create(emitter -> sqliteStorageAdapter.clear(emitter::onComplete, emitter::onError))
-                        .subscribeOn(Schedulers.io())
-                        .subscribe(onComplete::call,
-                            throwable -> onError.accept(new DataStoreException("Clear operation failed",
-                                    throwable, AmplifyException.REPORT_BUG_TO_AWS_SUGGESTION))),
-                onError);
+        stop(
+            () -> {
+                Disposable completable = Completable.create(
+                    emitter -> sqliteStorageAdapter.clear(emitter::onComplete, emitter::onError)
+                )
+                    .subscribeOn(Schedulers.io())
+                    .subscribe(onComplete::call,
+                        throwable -> onError.accept(new DataStoreException("Clear operation failed",
+                                                         throwable, AmplifyException.REPORT_BUG_TO_AWS_SUGGESTION)));
+                pluginDisposables.add(completable);
+            },
+            onError
+        );
     }
 
     /**

aws-datastore/src/main/java/com/amplifyframework/datastore/syncengine/Orchestrator.java

@@ -404,9 +404,12 @@ private void onApiSyncFailure(Throwable exception) {
             return;
         }
         LOG.warn("API sync failed - transitioning to LOCAL_ONLY.", exception);
-        Completable.fromAction(this::transitionToLocalOnly)
-            .doOnError(error -> LOG.warn("Transition to LOCAL_ONLY failed.", error))
-            .subscribe();
+        Disposable subscription = Completable.fromAction(this::transitionToLocalOnly)
+            .subscribe(
+                () -> { /* no-op */ },
+                error -> LOG.warn("Transition to LOCAL_ONLY failed.", error)
+            );
+        disposables.add(subscription);
     }
 
     private void disposeNetworkChanges() {
@@ -422,13 +425,16 @@ private void monitorNetworkChanges() {
         monitorNetworkChangesDisposable = reachabilityMonitor.getObservable()
             .skip(1) // We skip the current online state, we only care about transitions
             .filter(ignore -> !State.STOPPED.equals(currentState.get()))
-            .subscribe(isOnline -> {
-                if (isOnline) {
-                    transitionToApiSync();
-                } else {
-                    transitionToLocalOnly();
-                }
-            });
+            .subscribe(
+                isOnline -> {
+                    if (isOnline) {
+                        transitionToApiSync();
+                    } else {
+                        transitionToLocalOnly();
+                    }
+                },
+                error -> LOG.warn("Error observing network changes", error)
+            );
     }
 
     /**

aws-datastore/src/main/java/com/amplifyframework/datastore/syncengine/RetryStrategy.java

@@ -24,6 +24,7 @@
 import java.util.concurrent.TimeUnit;
 
 import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.schedulers.Schedulers;
 
 /**
  * Class that defines inner classes and interfaces related to retry strategies.
@@ -71,7 +72,11 @@ public boolean retryHandler(int attemptNumber, Throwable throwable) {
             } else {
                 final long waitTimeSeconds = Double.valueOf(Math.pow(2, attemptNumber % maxExponent)).longValue();
                 LOG.debug("Waiting " + waitTimeSeconds + " seconds before retrying");
-                Completable.timer(TimeUnit.SECONDS.toMillis(waitTimeSeconds), TimeUnit.MILLISECONDS).blockingAwait();
+                Completable.timer(
+                    TimeUnit.SECONDS.toMillis(waitTimeSeconds),
+                    TimeUnit.MILLISECONDS,
+                    Schedulers.io()
+                ).blockingAwait();
                 return true;
             }
         }

aws-datastore/src/main/java/com/amplifyframework/datastore/syncengine/SubscriptionProcessor.java

@@ -260,7 +260,10 @@ void startDrainingMutationBuffer() {
                 .flatMapCompletable(this::mergeEvent)
                 .doOnError(failure -> LOG.warn("Reading subscriptions buffer has failed.", failure))
                 .doOnComplete(() -> LOG.warn("Reading from subscriptions buffer is completed."))
-                .subscribe()
+                .subscribe(
+                    () -> LOG.info("Subscription data buffer processing complete"),
+                    error -> LOG.warn("Error draining subscription data buffer", error)
+                )
         );
     }