Remove synchronized blocks from ApkInstaller

Author: lfkellogg

firebase-appdistribution/src/main/java/com/google/firebase/appdistribution/impl/ApkInstaller.java

@@ -14,46 +14,35 @@
 
 package com.google.firebase.appdistribution.impl;
 
-import static com.google.firebase.appdistribution.impl.TaskUtils.safeSetTaskException;
-
 import android.app.Activity;
 import android.content.Intent;
-import androidx.annotation.GuardedBy;
 import androidx.annotation.Nullable;
 import com.google.android.gms.tasks.Task;
 import com.google.android.gms.tasks.TaskCompletionSource;
+import com.google.firebase.annotations.concurrent.Lightweight;
 import com.google.firebase.appdistribution.FirebaseAppDistribution;
 import com.google.firebase.appdistribution.FirebaseAppDistributionException;
+import java.util.concurrent.Executor;
 
 /** Class that handles installing APKs in {@link FirebaseAppDistribution}. */
 class ApkInstaller {
   private static final String TAG = "ApkInstaller";
 
   private final FirebaseAppDistributionLifecycleNotifier lifeCycleNotifier;
+  private final TaskCompletionSourceCache<Void> installTaskCompletionSourceCache;
+  private final @Lightweight Executor lightweightExecutor;
 
-  @GuardedBy("installTaskLock")
-  private TaskCompletionSource<Void> installTaskCompletionSource;
-
-  private final Object installTaskLock = new Object();
-
-  ApkInstaller(FirebaseAppDistributionLifecycleNotifier lifeCycleNotifier) {
+  ApkInstaller(
+      FirebaseAppDistributionLifecycleNotifier lifeCycleNotifier,
+      @Lightweight Executor lightweightExecutor) {
     this.lifeCycleNotifier = lifeCycleNotifier;
-    lifeCycleNotifier.addOnActivityStartedListener(this::onActivityStarted);
+    this.installTaskCompletionSourceCache = new TaskCompletionSourceCache<>(lightweightExecutor);
+    this.lightweightExecutor = lightweightExecutor;
     lifeCycleNotifier.addOnActivityDestroyedListener(this::onActivityDestroyed);
   }
 
-  ApkInstaller() {
-    this(FirebaseAppDistributionLifecycleNotifier.getInstance());
-  }
-
-  void onActivityStarted(@Nullable Activity activity) {
-    synchronized (installTaskLock) {
-      if (installTaskCompletionSource == null
-          || installTaskCompletionSource.getTask().isComplete()
-          || activity == null) {
-        return;
-      }
-    }
+  ApkInstaller(@Lightweight Executor lightweightExecutor) {
+    this(FirebaseAppDistributionLifecycleNotifier.getInstance(), lightweightExecutor);
   }
 
   void onActivityDestroyed(@Nullable Activity activity) {
@@ -65,15 +54,11 @@ void onActivityDestroyed(@Nullable Activity activity) {
   }
 
   Task<Void> installApk(String path, Activity currentActivity) {
-    synchronized (installTaskLock) {
-      startInstallActivity(path, currentActivity);
-
-      if (this.installTaskCompletionSource == null
-          || this.installTaskCompletionSource.getTask().isComplete()) {
-        this.installTaskCompletionSource = new TaskCompletionSource<>();
-      }
-      return installTaskCompletionSource.getTask();
-    }
+    return installTaskCompletionSourceCache.getOrCreateTaskFromCompletionSource(
+        () -> {
+          startInstallActivity(path, currentActivity);
+          return new TaskCompletionSource<>();
+        });
   }
 
   private void startInstallActivity(String path, Activity currentActivity) {
@@ -84,12 +69,9 @@ private void startInstallActivity(String path, Activity currentActivity) {
   }
 
   void trySetInstallTaskError() {
-    synchronized (installTaskLock) {
-      safeSetTaskException(
-          installTaskCompletionSource,
-          new FirebaseAppDistributionException(
-              ErrorMessages.APK_INSTALLATION_FAILED,
-              FirebaseAppDistributionException.Status.INSTALLATION_FAILURE));
-    }
+    installTaskCompletionSourceCache.setException(
+        new FirebaseAppDistributionException(
+            ErrorMessages.APK_INSTALLATION_FAILED,
+            FirebaseAppDistributionException.Status.INSTALLATION_FAILURE));
   }
 }

firebase-appdistribution/src/main/java/com/google/firebase/appdistribution/impl/FirebaseAppDistributionRegistrar.java

@@ -121,7 +121,7 @@ private FirebaseAppDistribution buildFirebaseAppDistribution(
                 firebaseApp, firebaseInstallationsApiProvider, signInStorage, blockingExecutor),
             new NewReleaseFetcher(
                 firebaseApp.getApplicationContext(), testerApiClient, releaseIdentifier),
-            new ApkUpdater(firebaseApp, new ApkInstaller(), blockingExecutor),
+            new ApkUpdater(firebaseApp, new ApkInstaller(lightweightExecutor), blockingExecutor),
             new AabUpdater(blockingExecutor),
             signInStorage,
             lifecycleNotifier,

firebase-appdistribution/src/main/java/com/google/firebase/appdistribution/impl/TaskCompletionSourceCache.java

@@ -0,0 +1,101 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.firebase.appdistribution.impl;
+
+import com.google.android.gms.tasks.Task;
+import com.google.android.gms.tasks.TaskCompletionSource;
+import com.google.firebase.annotations.concurrent.Lightweight;
+import com.google.firebase.concurrent.FirebaseExecutors;
+import java.util.concurrent.Executor;
+
+/**
+ * A cache for a {@link TaskCompletionSource}, for use in cases where we only ever want one active
+ * task at a time for a particular operation.
+ *
+ * <p>This is equivalent to {@link UpdateTaskCache} but for Tasks.
+ */
+class TaskCompletionSourceCache<T> {
+
+  /** A functional interface for a producer of a new {@link TaskCompletionSource}. */
+  @FunctionalInterface
+  interface TaskCompletionSourceProducer<T> {
+
+    /** Produce a new {@link TaskCompletionSource}. */
+    TaskCompletionSource<T> produce();
+  }
+
+  private TaskCompletionSource<T> cachedTaskCompletionSource;
+  private final Executor sequentialExecutor;
+
+  /**
+   * Constructor for a {@link TaskCompletionSourceCache} that controls access using its own
+   * sequential executor backed by the given base executor.
+   *
+   * @param baseExecutor Executor (typically {@link Lightweight}) to back the sequential executor.
+   */
+  TaskCompletionSourceCache(Executor baseExecutor) {
+    sequentialExecutor = FirebaseExecutors.newSequentialExecutor(baseExecutor);
+  }
+
+  /**
+   * Gets a cached {@link TaskCompletionSource}, if there is one and it is not completed, or else
+   * calls the given {@code producer} and caches the return value.
+   */
+  Task<T> getOrCreateTaskFromCompletionSource(TaskCompletionSourceProducer<T> producer) {
+    TaskCompletionSource<T> taskCompletionSource = new TaskCompletionSource<>();
+    sequentialExecutor.execute(
+        () -> {
+          if (!isOngoing(cachedTaskCompletionSource)) {
+            cachedTaskCompletionSource = producer.produce();
+          }
+          TaskUtils.shadowTask(taskCompletionSource, cachedTaskCompletionSource.getTask());
+        });
+    return taskCompletionSource.getTask();
+  }
+
+  /**
+   * Sets the result on the cached {@link TaskCompletionSource}, if there is one and it is not
+   * completed.
+   */
+  Task<Void> setResult(T result) {
+    TaskCompletionSource<Void> taskCompletionSource = new TaskCompletionSource<>();
+    sequentialExecutor.execute(
+        () -> {
+          if (isOngoing(cachedTaskCompletionSource)) {
+            cachedTaskCompletionSource.setResult(result);
+          }
+        });
+    return taskCompletionSource.getTask();
+  }
+
+  /**
+   * Sets the exception on the cached {@link TaskCompletionSource}, if there is one and it is not
+   * completed.
+   */
+  Task<Void> setException(Exception e) {
+    TaskCompletionSource<Void> taskCompletionSource = new TaskCompletionSource<>();
+    sequentialExecutor.execute(
+        () -> {
+          if (isOngoing(cachedTaskCompletionSource)) {
+            cachedTaskCompletionSource.setException(e);
+          }
+        });
+    return taskCompletionSource.getTask();
+  }
+
+  private static <T> boolean isOngoing(TaskCompletionSource<T> taskCompletionSource) {
+    return taskCompletionSource != null && !taskCompletionSource.getTask().isComplete();
+  }
+}

firebase-appdistribution/src/main/java/com/google/firebase/appdistribution/impl/TaskUtils.java

@@ -21,6 +21,7 @@
 import com.google.firebase.appdistribution.FirebaseAppDistributionException;
 import com.google.firebase.appdistribution.FirebaseAppDistributionException.Status;
 import com.google.firebase.appdistribution.UpdateTask;
+import com.google.firebase.concurrent.FirebaseExecutors;
 import java.util.concurrent.Executor;
 
 class TaskUtils {
@@ -144,5 +145,12 @@ static <T> UpdateTask onSuccessUpdateTask(
     return updateTask;
   }
 
+  /** Set a {@link TaskCompletionSource} to be resolved with the result of another {@link Task}. */
+  static <T> void shadowTask(TaskCompletionSource<T> taskCompletionSource, Task<T> task) {
+    task.addOnSuccessListener(FirebaseExecutors.directExecutor(), taskCompletionSource::setResult)
+        .addOnFailureListener(
+            FirebaseExecutors.directExecutor(), taskCompletionSource::setException);
+  }
+
   private TaskUtils() {}
 }

firebase-appdistribution/src/test/java/com/google/firebase/appdistribution/impl/ApkInstallerTests.java

@@ -14,14 +14,27 @@
 
 package com.google.firebase.appdistribution.impl;
 
+import static com.google.firebase.appdistribution.FirebaseAppDistributionException.Status.INSTALLATION_FAILURE;
+import static com.google.firebase.appdistribution.impl.TestUtils.awaitAsyncOperations;
+import static com.google.firebase.appdistribution.impl.TestUtils.awaitTaskFailure;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import static org.robolectric.Shadows.shadowOf;
 
+import android.app.Activity;
 import android.content.Intent;
 import com.google.android.gms.tasks.Task;
+import com.google.firebase.annotations.concurrent.Lightweight;
 import com.google.firebase.appdistribution.FirebaseAppDistributionException;
 import com.google.firebase.appdistribution.impl.FirebaseAppDistributionServiceImplTest.TestActivity;
+import com.google.firebase.concurrent.TestOnlyExecutors;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -37,58 +50,52 @@ public class ApkInstallerTests {
   private ShadowActivity shadowActivity;
   private ApkInstaller apkInstaller;
   @Mock private FirebaseAppDistributionLifecycleNotifier mockLifecycleNotifier;
+  @Lightweight private ExecutorService lightweightExecutor;
 
   @Before
   public void setUp() {
     MockitoAnnotations.initMocks(this);
 
     activity = Robolectric.buildActivity(TestActivity.class).create().get();
     shadowActivity = shadowOf(activity);
-    apkInstaller = new ApkInstaller(mockLifecycleNotifier);
+    lightweightExecutor = TestOnlyExecutors.lite();
+    apkInstaller = new ApkInstaller(mockLifecycleNotifier, lightweightExecutor);
   }
 
   @Test
-  public void installApk_currentActivityNotNull_InstallIntentOnCurrentActivity() {
+  public void installApk_currentActivityNotNull_InstallIntentOnCurrentActivity()
+      throws InterruptedException {
     String path = "path";
     Task<Void> installTask = apkInstaller.installApk(path, activity);
+    awaitAsyncOperations(lightweightExecutor);
     Intent installIntent = shadowActivity.getNextStartedActivity();
 
     assertFalse(installTask.isComplete());
     assertEquals(
         "com.google.firebase.appdistribution.impl.InstallActivity",
         installIntent.getComponent().getShortClassName());
     assertEquals(path, installIntent.getExtras().get("INSTALL_PATH"));
-  }
-
-  @Test
-  public void
-      setCurrentActivity_appInForegroundAfterAnInstallAttempt_installIntentOnCurrentActivity() {
-    apkInstaller.onActivityStarted(null);
-    String path = "path123";
-    Task<Void> installTask = apkInstaller.installApk(path, activity);
-    apkInstaller.onActivityStarted(activity);
-
-    Intent installIntent = shadowActivity.getNextStartedActivity();
-
-    assertEquals(
-        "com.google.firebase.appdistribution.impl.InstallActivity",
-        installIntent.getComponent().getShortClassName());
-    assertEquals(path, installIntent.getExtras().get("INSTALL_PATH"));
     assertFalse(installTask.isComplete());
   }
 
   @Test
   public void installActivityDestroyed_setsInstallError() {
     Task<Void> installTask = apkInstaller.installApk("path", activity);
     apkInstaller.onActivityDestroyed(new InstallActivity());
-    Exception ex = installTask.getException();
 
-    assert ex instanceof FirebaseAppDistributionException;
-    assertEquals(
-        ErrorMessages.APK_INSTALLATION_FAILED,
-        ((FirebaseAppDistributionException) ex).getMessage());
-    assertEquals(
-        FirebaseAppDistributionException.Status.INSTALLATION_FAILURE,
-        ((FirebaseAppDistributionException) ex).getErrorCode());
+    awaitTaskFailure(installTask, INSTALLATION_FAILURE, ErrorMessages.APK_INSTALLATION_FAILED);
+  }
+
+  @Test
+  public void whenCalledMultipleTimes_onlyEmitsOneIntent()
+      throws FirebaseAppDistributionException, ExecutionException, InterruptedException {
+    Activity mockActivity = mock(Activity.class);
+    doNothing().when(mockActivity).startActivity(any());
+
+    apkInstaller.installApk("path", mockActivity);
+    apkInstaller.installApk("path", mockActivity);
+    awaitAsyncOperations(lightweightExecutor);
+
+    verify(mockActivity, times(1)).startActivity(any());
   }
 }