Make enqueue() return a Task<>. (#175)

* Make enqueue(Runnable) return Task.
* Add enqueue(Callable) which returns Task.
* Add enqueuAndForget(Runnable) which returns void.
* Update usages to make use of new ability (more closely matching TypeScript).

Author: mikelehen

firebase-firestore/src/androidTest/java/com/google/firebase/firestore/remote/StreamTest.java

@@ -89,9 +89,9 @@ public void onWriteResponse(
   /** Puts the write in a state that accepts mutations. */
   private StreamStatusCallback waitForOpenStream(AsyncQueue testQueue, WriteStream writeStream) {
     StreamStatusCallback callback = new StreamStatusCallback();
-    testQueue.enqueue(() -> writeStream.start(callback));
+    testQueue.enqueueAndForget(() -> writeStream.start(callback));
     waitFor(callback.openSemaphore);
-    testQueue.enqueue(writeStream::writeHandshake);
+    testQueue.enqueueAndForget(writeStream::writeHandshake);
     waitFor(callback.handshakeSemaphore);
     return callback;
   }
@@ -105,7 +105,7 @@ public void testWatchStreamStopBeforeHandshake() throws Exception {
     final WatchStream watchStream = datastore.createWatchStream();
     StreamStatusCallback streamCallback = new StreamStatusCallback() {};
 
-    testQueue.enqueue(() -> watchStream.start(streamCallback));
+    testQueue.enqueueAndForget(() -> watchStream.start(streamCallback));
     waitFor(streamCallback.openSemaphore);
 
     // Stop must not call watchStreamStreamDidClose because the full implementation of the delegate
@@ -137,21 +137,21 @@ public void onWriteResponse(
             super.onWriteResponse(commitVersion, mutationResults);
           }
         };
-    testQueue.enqueue(() -> writeStream.start(streamCallback));
+    testQueue.enqueueAndForget(() -> writeStream.start(streamCallback));
     waitFor(streamCallback.openSemaphore);
 
     // Writing before the handshake should throw
-    testQueue.enqueue(
+    testQueue.enqueueAndForget(
         () -> {
           assertThrows(Throwable.class, () -> writeStream.writeMutations(mutations));
         });
 
     // Handshake should always be called
-    testQueue.enqueue(writeStream::writeHandshake);
+    testQueue.enqueueAndForget(writeStream::writeHandshake);
     waitFor(streamCallback.handshakeSemaphore);
 
     // Now writes should succeed
-    testQueue.enqueue(() -> writeStream.writeMutations(mutations));
+    testQueue.enqueueAndForget(() -> writeStream.writeMutations(mutations));
     waitFor(streamCallback.responseReceivedSemaphore);
 
     testQueue.runSync(writeStream::stop);
@@ -167,7 +167,7 @@ public void testWriteStreamStopPartial() throws Exception {
     final WriteStream writeStream = datastore.createWriteStream();
 
     StreamStatusCallback streamCallback = new StreamStatusCallback() {};
-    testQueue.enqueue(() -> writeStream.start(streamCallback));
+    testQueue.enqueueAndForget(() -> writeStream.start(streamCallback));
     waitFor(streamCallback.openSemaphore);
 
     // Don't start the handshake
@@ -203,7 +203,7 @@ public void testStreamClosesWhenIdle() throws Exception {
     final WriteStream writeStream = datastore.createWriteStream();
     StreamStatusCallback callback = waitForOpenStream(testQueue, writeStream);
 
-    testQueue.enqueue(
+    testQueue.enqueueAndForget(
         () -> {
           writeStream.markIdle();
           assertTrue(testQueue.containsDelayedTask(TimerId.WRITE_STREAM_IDLE));

firebase-firestore/src/main/java/com/google/firebase/firestore/FirebaseFirestore.java

@@ -80,7 +80,7 @@ static FirebaseFirestore newInstance(
 
     CredentialsProvider provider = new FirebaseAuthCredentialsProvider(app);
 
-    queue.enqueue(
+    queue.enqueueAndForget(
         new Runnable() {
           @Override
           public void run() {

firebase-firestore/src/main/java/com/google/firebase/firestore/core/FirestoreClient.java

@@ -25,7 +25,6 @@
 import com.google.firebase.firestore.local.SQLitePersistence;
 import com.google.firebase.firestore.model.Document;
 import com.google.firebase.firestore.model.DocumentKey;
-import com.google.firebase.firestore.model.MaybeDocument;
 import com.google.firebase.firestore.model.mutation.Mutation;
 import com.google.firebase.firestore.model.mutation.MutationBatchResult;
 import com.google.firebase.firestore.remote.Datastore;
@@ -73,14 +72,14 @@ public FirestoreClient(
             hardAssert(!firstUser.getTask().isComplete(), "Already fulfilled first user task");
             firstUser.setResult(user);
           } else {
-            asyncQueue.enqueue(() -> syncEngine.handleUserChange(user));
+            asyncQueue.enqueueAndForget(() -> syncEngine.handleUserChange(user));
           }
         });
 
     // Defer initialization until we get the current user from the userChangeListener. This is
     // guaranteed to be synchronously dispatched onto our worker queue, so we will be initialized
     // before any subsequently queued work runs.
-    asyncQueue.enqueue(
+    asyncQueue.enqueueAndForget(
         () -> {
           try {
             // Block on initial user being available
@@ -93,74 +92,58 @@ public FirestoreClient(
   }
 
   public Task<Void> disableNetwork() {
-    final TaskCompletionSource<Void> source = new TaskCompletionSource<>();
-    asyncQueue.enqueue(
-        () -> {
-          remoteStore.disableNetwork();
-          source.setResult(null);
-        });
-    return source.getTask();
+    return asyncQueue.enqueue(() -> remoteStore.disableNetwork());
   }
 
   public Task<Void> enableNetwork() {
-    final TaskCompletionSource<Void> source = new TaskCompletionSource<>();
-    asyncQueue.enqueue(
-        () -> {
-          remoteStore.enableNetwork();
-          source.setResult(null);
-        });
-    return source.getTask();
+    return asyncQueue.enqueue(() -> remoteStore.enableNetwork());
   }
 
   /** Shuts down this client, cancels all writes / listeners, and releases all resources. */
   public Task<Void> shutdown() {
-    final TaskCompletionSource<Void> source = new TaskCompletionSource<>();
     credentialsProvider.removeUserChangeListener();
-    asyncQueue.enqueue(
+    return asyncQueue.enqueue(
         () -> {
           remoteStore.shutdown();
           persistence.shutdown();
-          source.setResult(null);
         });
-    return source.getTask();
   }
 
   /** Starts listening to a query. */
   public QueryListener listen(
       Query query, ListenOptions options, EventListener<ViewSnapshot> listener) {
     QueryListener queryListener = new QueryListener(query, options, listener);
-    asyncQueue.enqueue(() -> eventManager.addQueryListener(queryListener));
+    asyncQueue.enqueueAndForget(() -> eventManager.addQueryListener(queryListener));
     return queryListener;
   }
 
   /** Stops listening to a query previously listened to. */
   public void stopListening(QueryListener listener) {
-    asyncQueue.enqueue(() -> eventManager.removeQueryListener(listener));
+    asyncQueue.enqueueAndForget(() -> eventManager.removeQueryListener(listener));
   }
 
   public Task<Document> getDocumentFromLocalCache(DocumentKey docKey) {
-    final TaskCompletionSource<Document> source = new TaskCompletionSource<>();
-    asyncQueue.enqueue(
-        () -> {
-          MaybeDocument maybeDoc = localStore.readDocument(docKey);
-          if (maybeDoc instanceof Document) {
-            source.setResult((Document) maybeDoc);
-          } else {
-            source.setException(
-                new FirebaseFirestoreException(
+    return asyncQueue
+        .enqueue(
+            () -> {
+              return localStore.readDocument(docKey);
+            })
+        .continueWith(
+            (maybeDoc) -> {
+              if (maybeDoc.getResult() instanceof Document) {
+                return (Document) maybeDoc.getResult();
+              } else {
+                throw new FirebaseFirestoreException(
                     "Failed to get document from cache. (However, this document may exist on the "
                         + "server. Run again without setting source to CACHE to attempt "
                         + "to retrieve the document from the server.)",
-                    Code.UNAVAILABLE));
-          }
-        });
-
-    return source.getTask();
+                    Code.UNAVAILABLE);
+              }
+            });
   }
 
   public Task<ViewSnapshot> getDocumentsFromLocalCache(Query query) {
-    final TaskCompletionSource<ViewSnapshot> source = new TaskCompletionSource<>();
-    asyncQueue.enqueue(
+    return asyncQueue.enqueue(
         () -> {
           ImmutableSortedMap<DocumentKey, Document> docs = localStore.executeQuery(query);
 
@@ -170,16 +153,14 @@ public Task<ViewSnapshot> getDocumentsFromLocalCache(Query query) {
                   new ImmutableSortedSet<DocumentKey>(
                       Collections.emptyList(), DocumentKey::compareTo));
           View.DocumentChanges viewDocChanges = view.computeDocChanges(docs);
-          source.setResult(view.applyChanges(viewDocChanges).getSnapshot());
+          return view.applyChanges(viewDocChanges).getSnapshot();
         });
-
-    return source.getTask();
   }
 
   /** Writes mutations. The returned task will be notified when it's written to the backend. */
   public Task<Void> write(final List<Mutation> mutations) {
     final TaskCompletionSource<Void> source = new TaskCompletionSource<>();
-    asyncQueue.enqueue(() -> syncEngine.writeMutations(mutations, source));
+    asyncQueue.enqueueAndForget(() -> syncEngine.writeMutations(mutations, source));
     return source.getTask();
   }
 

firebase-firestore/src/main/java/com/google/firebase/firestore/remote/AbstractStream.java

@@ -197,7 +197,7 @@ public void start(CallbackT listener) {
     state = State.Auth;
 
     // Make sure onOpen call is asynchronously
-    workerQueue.enqueue(
+    workerQueue.enqueueAndForget(
         () -> {
           // The stream could have been stopped between now and scheduling the onOpen.
           // Alternatively, onError could have been called (and the listener removed) before

firebase-firestore/src/main/java/com/google/firebase/firestore/util/AsyncQueue.java

@@ -26,6 +26,7 @@
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import javax.annotation.CheckReturnValue;
 
 /** A helper class that allows to schedule/queue Runnables on a single threaded background queue. */
 public class AsyncQueue {
@@ -256,18 +257,58 @@ public void verifyIsCurrentThread() {
   }
 
   /**
-   * Queue and run this task immediately after every other already queued task.
+   * Queue and run this Callable task immediately after every other already queued task.
    *
    * @param task The task to run.
+   * @return A Task object that is resolved after the enqueued operation has completed, with the
+   *     return value of the operation.
    */
-  public void enqueue(Runnable task) {
+  @CheckReturnValue
+  public <T> Task<T> enqueue(Callable<T> task) {
+    final TaskCompletionSource<T> completionSource = new TaskCompletionSource<>();
     try {
-      executor.execute(task);
+      executor.execute(
+          () -> {
+            try {
+              completionSource.setResult(task.call());
+            } catch (Exception e) {
+              completionSource.setException(e);
+              throw new RuntimeException(e);
+            }
+          });
     } catch (RejectedExecutionException e) {
       // The only way we can get here is if the AsyncQueue has panicked and we're now racing with
       // the post to the main looper that will crash the app.
       Logger.warn(AsyncQueue.class.getSimpleName(), "Refused to enqueue task after panic");
     }
+    return completionSource.getTask();
+  }
+
+  /**
+   * Queue and run this Runnable task immediately after every other already queued task.
+   *
+   * @param task The task to run.
+   * @return A Task object that is resolved after the enqueued operation has completed.
+   */
+  @CheckReturnValue
+  public Task<Void> enqueue(Runnable task) {
+    return enqueue(
+        () -> {
+          task.run();
+          return null;
+        });
+  }
+
+  /**
+   * Queue and run this Runnable task immediately after every other already queued task. Unlike
+   * enqueue(), returns void instead of a Task Object for use when we have no need to "wait" on the
+   * task completing.
+   *
+   * @param task The task to run.
+   */
+  @SuppressWarnings("CheckReturnValue")
+  public void enqueueAndForget(Runnable task) {
+    enqueue(task);
   }
 
   /**
@@ -331,7 +372,7 @@ public void panic(Throwable t) {
   @VisibleForTesting
   public void runSync(Runnable task) throws InterruptedException {
     Semaphore done = new Semaphore(0);
-    enqueue(
+    enqueueAndForget(
         () -> {
           task.run();
           done.release();

firebase-firestore/src/test/java/com/google/firebase/firestore/util/AsyncQueueTest.java

@@ -62,9 +62,9 @@ private void waitForExpectedSteps() {
   @Test
   public void canScheduleTasksInTheFuture() {
     expectedSteps = Arrays.asList(1, 2, 3, 4);
-    queue.enqueue(runnableForStep(1));
+    queue.enqueueAndForget(runnableForStep(1));
     queue.enqueueAfterDelay(TIMER_ID_1, 5, runnableForStep(4));
-    queue.enqueue(runnableForStep(2));
+    queue.enqueueAndForget(runnableForStep(2));
     queue.enqueueAfterDelay(TIMER_ID_2, 1, runnableForStep(3));
     waitForExpectedSteps();
   }
@@ -73,9 +73,9 @@ public void canScheduleTasksInTheFuture() {
   public void canCancelDelayedTasks() {
     expectedSteps = Arrays.asList(1, 3);
     // Queue everything from the queue to ensure nothing completes before we cancel.
-    queue.enqueue(
+    queue.enqueueAndForget(
         () -> {
-          queue.enqueue(runnableForStep(1));
+          queue.enqueueAndForget(runnableForStep(1));
           DelayedTask step2Timer = queue.enqueueAfterDelay(TIMER_ID_1, 1, runnableForStep(2));
           queue.enqueueAfterDelay(TIMER_ID_3, 5, runnableForStep(3));
 
@@ -89,22 +89,22 @@ public void canCancelDelayedTasks() {
 
   @Test
   public void canManuallyDrainAllDelayedTasksForTesting() throws Exception {
-    queue.enqueue(runnableForStep(1));
+    queue.enqueueAndForget(runnableForStep(1));
     queue.enqueueAfterDelay(TIMER_ID_1, 20, runnableForStep(4));
     queue.enqueueAfterDelay(TIMER_ID_2, 10, runnableForStep(3));
-    queue.enqueue(runnableForStep(2));
+    queue.enqueueAndForget(runnableForStep(2));
 
     queue.runDelayedTasksUntil(TimerId.ALL);
     assertEquals(Arrays.asList(1, 2, 3, 4), completedSteps);
   }
 
   @Test
   public void canManuallyDrainSpecificDelayedTasksForTesting() throws Exception {
-    queue.enqueue(runnableForStep(1));
+    queue.enqueueAndForget(runnableForStep(1));
     queue.enqueueAfterDelay(TIMER_ID_1, 20, runnableForStep(5));
     queue.enqueueAfterDelay(TIMER_ID_2, 10, runnableForStep(3));
     queue.enqueueAfterDelay(TIMER_ID_3, 15, runnableForStep(4));
-    queue.enqueue(runnableForStep(2));
+    queue.enqueueAndForget(runnableForStep(2));
 
     queue.runDelayedTasksUntil(TIMER_ID_3);
     assertEquals(Arrays.asList(1, 2, 3, 4), completedSteps);