Review feedback, connected check

Author: jmwski

firebase-database/src/androidTest/java/com/google/firebase/database/QueryTest.java

@@ -3424,8 +3424,7 @@ public void emptyQueryGet() throws DatabaseException, InterruptedException {
         appForDatabaseUrl(IntegrationTestValues.getNamespace(), UUID.randomUUID().toString());
     FirebaseDatabase db = FirebaseDatabase.getInstance(app);
     try {
-      Tasks.await(db.getReference("dummy/").setValue(42L));
-      assertNull(Tasks.await(db.getReference("null/").get()).getValue());
+      assertNull(Tasks.await(db.getReference(UUID.randomUUID().toString()).get()).getValue());
     } catch (ExecutionException e) {
       fail(e.getMessage());
     }
@@ -3436,12 +3435,8 @@ public void offlineQueryGet() throws DatabaseException, InterruptedException {
     FirebaseApp app =
         appForDatabaseUrl(IntegrationTestValues.getNamespace(), UUID.randomUUID().toString());
     FirebaseDatabase db = FirebaseDatabase.getInstance(app);
+    db.setLogLevel(Logger.Level.DEBUG);
     DatabaseReference node = db.getReference();
-    try {
-      Tasks.await(node.setValue(42L));
-    } catch (ExecutionException e) {
-      fail();
-    }
     db.goOffline();
     try {
       Tasks.await(node.get());
@@ -3453,22 +3448,19 @@ public void offlineQueryGet() throws DatabaseException, InterruptedException {
   }
 
   @Test
-  public void getQueryBasic() throws DatabaseException, InterruptedException {
+  public void getQueryBasic() throws DatabaseException, InterruptedException, ExecutionException {
     FirebaseApp app =
         appForDatabaseUrl(IntegrationTestValues.getNamespace(), UUID.randomUUID().toString());
     FirebaseDatabase db = FirebaseDatabase.getInstance(app);
     DatabaseReference node = db.getReference();
-    try {
-      Tasks.await(node.setValue(42));
-      assertEquals(42L, Tasks.await(node.get()).getValue());
-    } catch (ExecutionException e) {
-      fail(e.getMessage());
-    }
+    Tasks.await(node.setValue(42));
+    assertEquals(42L, Tasks.await(node.get()).getValue());
   }
 
   @Test
   public void getQueryCached()
-      throws DatabaseException, InterruptedException, TimeoutException, TestFailure {
+      throws DatabaseException, InterruptedException, TimeoutException, TestFailure,
+          ExecutionException {
     FirebaseApp app =
         appForDatabaseUrl(IntegrationTestValues.getAltNamespace(), UUID.randomUUID().toString());
     FirebaseDatabase db = FirebaseDatabase.getInstance(app);
@@ -3493,15 +3485,13 @@ public void onCancelled(@NonNull DatabaseError error) {}
     try {
       // Since we still have a listener on `ref`, the 42L should be cached here.
       assertEquals(42L, Tasks.await(ref.get()).getValue());
-    } catch (ExecutionException e) {
-      fail(e.getMessage());
     } finally {
       ref.removeEventListener(listener);
     }
   }
 
   @Test
-  public void getQuerySkipsCacheWhenOnline()
+  public void getRetrievesLatestServerValue()
       throws DatabaseException, InterruptedException, ExecutionException, TestFailure,
           TimeoutException {
     FirebaseApp readerApp =
@@ -3513,13 +3503,13 @@ public void getQuerySkipsCacheWhenOnline()
     DatabaseReference reader = readerDb.getReference();
     DatabaseReference writer = writerDb.getReference();
 
-    final Semaphore semaphore = new Semaphore(0);
+    final Semaphore readerSemaphore = new Semaphore(0);
     reader.addValueEventListener(
         new ValueEventListener() {
           @Override
           public void onDataChange(@NonNull DataSnapshot snapshot) {
             if (snapshot.getValue() != null && snapshot.getValue().equals(42L)) {
-              semaphore.release();
+              readerSemaphore.release();
             }
           }
 
@@ -3529,7 +3519,7 @@ public void onCancelled(@NonNull DatabaseError error) {}
 
     WriteFuture write = new WriteFuture(writer, 42L);
     assertNull(write.timedGet());
-    IntegrationTestHelpers.waitFor(semaphore);
+    IntegrationTestHelpers.waitFor(readerSemaphore);
 
     write = new WriteFuture(writer, 43L);
     assertNull(write.timedGet());
@@ -3541,6 +3531,41 @@ public void onCancelled(@NonNull DatabaseError error) {}
     }
   }
 
+  @Test
+  public void getUpdatesPersistenceCacheWhenEnabled()
+      throws DatabaseException, InterruptedException, ExecutionException, TestFailure,
+          TimeoutException {
+    FirebaseApp readerApp =
+        appForDatabaseUrl(IntegrationTestValues.getNamespace(), UUID.randomUUID().toString());
+    FirebaseApp writerApp =
+        appForDatabaseUrl(IntegrationTestValues.getNamespace(), UUID.randomUUID().toString());
+    FirebaseDatabase readerDb = FirebaseDatabase.getInstance(readerApp);
+    readerDb.setPersistenceEnabled(true);
+    FirebaseDatabase writerDb = FirebaseDatabase.getInstance(writerApp);
+    DatabaseReference reader = readerDb.getReference();
+    DatabaseReference writer = writerDb.getReference();
+
+    assertNull(new WriteFuture(writer, 42L).timedGet());
+    assertEquals(42L, Tasks.await(reader.get()).getValue());
+
+    readerDb.goOffline();
+
+    Semaphore semaphore = new Semaphore(0);
+    reader.addListenerForSingleValueEvent(
+        new ValueEventListener() {
+          @Override
+          public void onDataChange(@NonNull DataSnapshot snapshot) {
+            if (snapshot.getValue() != null && snapshot.getValue().equals(42L)) {
+              semaphore.release();
+            }
+          }
+
+          @Override
+          public void onCancelled(@NonNull DatabaseError error) {}
+        });
+    IntegrationTestHelpers.waitFor(semaphore);
+  }
+
   @Test
   public void querySnapshotChildrenRespectDefaultOrdering()
       throws DatabaseException, ExecutionException, TimeoutException, TestFailure,

firebase-database/src/main/java/com/google/firebase/database/connection/PersistentConnectionImpl.java

@@ -29,9 +29,11 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 public class PersistentConnectionImpl implements Connection.Delegate, PersistentConnection {
 
@@ -111,6 +113,41 @@ public String toString() {
     }
   }
 
+  private static class OutstandingGet {
+    private Map<String, Object> request;
+    private ConnectionRequestCallback onComplete;
+    private String action;
+    private AtomicBoolean sent;
+
+    private OutstandingGet(
+        String action, Map<String, Object> request, ConnectionRequestCallback onComplete) {
+      this.action = action;
+      this.request = request;
+      this.onComplete = onComplete;
+      this.sent = new AtomicBoolean(false);
+    }
+
+    private String getAction() {
+      return action;
+    }
+
+    private ConnectionRequestCallback getOnComplete() {
+      return onComplete;
+    }
+
+    private Map<String, Object> getRequest() {
+      return request;
+    }
+
+    private boolean markSent() {
+      return sent.compareAndSet(false, true);
+    }
+
+    private boolean wasSent() {
+      return sent.get();
+    }
+  }
+
   private static class OutstandingPut {
     private String action;
     private Map<String, Object> request;
@@ -234,6 +271,7 @@ private enum ConnectionState {
   private static final long SUCCESSFUL_CONNECTION_ESTABLISHED_DELAY = 30 * 1000;
 
   private static final long IDLE_TIMEOUT = 60 * 1000;
+  private static final long GET_CONNECT_TIMEOUT = 3 * 1000;
 
   /** If auth fails repeatedly, we'll assume something is wrong and log a warning / back off. */
   private static final long INVALID_AUTH_TOKEN_THRESHOLD = 3;
@@ -253,11 +291,13 @@ private enum ConnectionState {
   private Connection realtime;
   private ConnectionState connectionState = ConnectionState.Disconnected;
   private long writeCounter = 0;
+  private long readCounter = 0;
   private long requestCounter = 0;
   private Map<Long, ConnectionRequestCallback> requestCBHash;
 
   private List<OutstandingDisconnect> onDisconnectRequestQueue;
   private Map<Long, OutstandingPut> outstandingPuts;
+  private Map<Long, OutstandingGet> outstandingGets;
 
   private Map<QuerySpec, OutstandingListen> listens;
   private String authToken;
@@ -287,6 +327,7 @@ public PersistentConnectionImpl(
     this.listens = new HashMap<QuerySpec, OutstandingListen>();
     this.requestCBHash = new HashMap<Long, ConnectionRequestCallback>();
     this.outstandingPuts = new HashMap<Long, OutstandingPut>();
+    this.outstandingGets = new ConcurrentHashMap<Long, OutstandingGet>();
     this.onDisconnectRequestQueue = new ArrayList<OutstandingDisconnect>();
     this.retryHelper =
         new RetryHelper.Builder(this.executorService, context.getLogger(), "ConnectionRetryHelper")
@@ -351,15 +392,58 @@ public void listen(
   public Task<Object> get(List<String> path, Map<String, Object> queryParams) {
     QuerySpec query = new QuerySpec(path, queryParams);
     TaskCompletionSource<Object> source = new TaskCompletionSource<>();
-    Task<Object> task;
-    if (connected()) {
-      task = sendGet(query, source);
-    } else {
-      source.setException(new Exception("Client is offline"));
-      task = source.getTask();
+
+    long readId = this.readCounter++;
+
+    Map<String, Object> request = new HashMap<String, Object>();
+    request.put(REQUEST_PATH, ConnectionUtils.pathToString(query.path));
+    request.put(REQUEST_QUERIES, query.queryParams);
+
+    outstandingGets.put(
+        readId,
+        new OutstandingGet(
+            REQUEST_ACTION_GET,
+            request,
+            new ConnectionRequestCallback() {
+              @Override
+              public void onResponse(Map<String, Object> response) {
+                String status = (String) response.get(REQUEST_STATUS);
+                if (status.equals("ok")) {
+                  Object body = response.get(SERVER_DATA_UPDATE_BODY);
+                  delegate.onDataUpdate(query.path, body, /*isMerge=*/ false, /*tagNumber=*/ null);
+                  source.setResult(body);
+                } else {
+                  source.setException(
+                      new Exception((String) response.get(SERVER_DATA_UPDATE_BODY)));
+                }
+              }
+            }));
+
+    if (!connected()) {
+      executorService.schedule(
+          new Runnable() {
+            @Override
+            public void run() {
+              OutstandingGet get = outstandingGets.get(readId);
+              if (get == null || !get.markSent()) {
+                return;
+              }
+              if (logger.logsDebug()) {
+                logger.debug("get " + readId + " timed out waiting for connection");
+              }
+              outstandingGets.remove(readId);
+              source.setException(new Exception("Client is offline"));
+            }
+          },
+          GET_CONNECT_TIMEOUT,
+          TimeUnit.MILLISECONDS);
+    }
+
+    if (canSendReads()) {
+      sendGet(readId);
     }
     doIdleCheck();
-    return task;
+    return source.getTask();
   }
 
   @Override
@@ -508,6 +592,10 @@ private boolean canSendWrites() {
     return connectionState == ConnectionState.Connected;
   }
 
+  private boolean canSendReads() {
+    return connectionState == ConnectionState.Connected;
+  }
+
   @Override
   public void onDisconnectMerge(
       List<String> path, Map<String, Object> updates, final RequestResultCallback onComplete) {
@@ -979,6 +1067,13 @@ private void restoreState() {
       sendPut(put);
     }
 
+    if (logger.logsDebug()) logger.debug("Restoring reads.");
+    ArrayList<Long> outstandingGetKeys = new ArrayList<Long>(outstandingGets.keySet());
+    Collections.sort(outstandingGetKeys);
+    for (Long getId : outstandingGetKeys) {
+      sendGet(getId);
+    }
+
     // Restore disconnect operations
     for (OutstandingDisconnect disconnect : onDisconnectRequestQueue) {
       sendOnDisconnect(
@@ -1067,27 +1162,32 @@ public void onResponse(Map<String, Object> response) {
         });
   }
 
-  private Task<Object> sendGet(final QuerySpec query, TaskCompletionSource<Object> source) {
-    Map<String, Object> request = new HashMap<String, Object>();
-    request.put(REQUEST_PATH, ConnectionUtils.pathToString(query.path));
-    request.put(REQUEST_QUERIES, query.queryParams);
+  private void sendGet(final Long readId) {
+    hardAssert(canSendReads(), "sendGet called when we can't send gets");
+    OutstandingGet get = outstandingGets.get(readId);
+    if (!get.markSent()) {
+      if (logger.logsDebug()) {
+        logger.debug("get" + readId + " already sent or cancelled, ignoring.");
+        return;
+      }
+    }
     sendAction(
-        REQUEST_ACTION_GET,
-        request,
+        get.getAction(),
+        get.getRequest(),
         new ConnectionRequestCallback() {
           @Override
           public void onResponse(Map<String, Object> response) {
-            String status = (String) response.get(REQUEST_STATUS);
-            if (status.equals("ok")) {
-              Object body = response.get(SERVER_DATA_UPDATE_BODY);
-              delegate.onDataUpdate(query.path, body, /*isMerge=*/ false, /*tagNumber=*/ null);
-              source.setResult(body);
+            OutstandingGet currentGet = outstandingGets.get(readId);
+            if (currentGet == get) {
+              outstandingGets.remove(readId);
+              get.getOnComplete().onResponse(response);
             } else {
-              source.setException(new Exception((String) response.get(SERVER_DATA_UPDATE_BODY)));
+              if (logger.logsDebug())
+                logger.debug(
+                    "Ignoring on complete for get " + readId + " because it was removed already.");
             }
           }
         });
-    return source.getTask();
   }
 
   private void sendListen(final OutstandingListen listen) {