Stronger Shutdown Semantics in Database (#90)

* Implementing stronger shutdown semantics for database (specifically RunLoop)

* Handling the case where the DB is not fully initialized

* Removing test file

* Running destroy and onClosed on RunLoop

Author: hiranya911

src/main/java/com/google/firebase/database/FirebaseDatabase.java

@@ -358,12 +358,7 @@ void destroy() {
       if (destroyed.get()) {
         return;
       }
-
-      if (repo != null) {
-        RepoManager.interrupt(repo);
-        repo = null;
-      }
-      RepoManager.interrupt(getConfig());
+      RepoManager.destroy(getConfig());
       destroyed.compareAndSet(false, true);
     }
   }

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

@@ -283,7 +283,11 @@ public void onDisconnect(Connection.DisconnectReason reason) {
     this.connectionState = ConnectionState.Disconnected;
     this.realtime = null;
     this.hasOnDisconnects = false;
-    requestCBHash.clear();
+    if (inactivityTimer != null) {
+      logger.debug("cancelling idle time checker");
+      inactivityTimer.cancel(false);
+      inactivityTimer = null;
+    }
     cancelSentTransactions();
     if (shouldReconnect()) {
       long timeSinceLastConnectSucceeded =

src/main/java/com/google/firebase/database/connection/WebsocketConnection.java

@@ -94,7 +94,7 @@ private WSClient createConnection(
     return new WSClientTubesock(ws);
   }
 
-  public void open() {
+  void open() {
     conn.connect();
     connectTimeout =
         executorService.schedule(
@@ -126,6 +126,7 @@ public void close() {
     }
     if (keepAlive != null) {
       keepAlive.cancel(true);
+      keepAlive = null;
     }
   }
 
@@ -139,8 +140,8 @@ public void send(Map<String, Object> message) {
         conn.send("" + segs.length);
       }
 
-      for (int i = 0; i < segs.length; ++i) {
-        conn.send(segs[i]);
+      for (String seg : segs) {
+        conn.send(seg);
       }
     } catch (IOException e) {
       logger.error("Failed to serialize message: " + message.toString(), e);
@@ -156,7 +157,6 @@ private void appendFrame(String message) {
       try {
         frameReader.freeze();
         Map<String, Object> decoded = JsonMapper.parseJson(frameReader.toString());
-        frameReader = null;
         if (logger.logsDebug()) {
           logger.debug("handleIncomingFrame complete frame: " + decoded);
         }
@@ -169,6 +169,8 @@ private void appendFrame(String message) {
         logger.error("Error parsing frame (cast error): " + frameReader.toString(), e);
         close();
         shutdown();
+      } finally {
+        frameReader = null;
       }
     }
   }
@@ -221,10 +223,8 @@ private void resetKeepAlive() {
         if (logger.logsDebug()) {
           logger.debug("Reset keepAlive. Remaining: " + keepAlive.getDelay(TimeUnit.MILLISECONDS));
         }
-      } else {
-        if (logger.logsDebug()) {
-          logger.debug("Reset keepAlive");
-        }
+      } else if (logger.logsDebug()) {
+        logger.debug("Reset keepAlive");
       }
       keepAlive = executorService.schedule(nop(), KEEP_ALIVE_TIMEOUT_MS, TimeUnit.MILLISECONDS);
     }
@@ -256,6 +256,7 @@ private void onClosed() {
     conn = null;
     if (keepAlive != null) {
       keepAlive.cancel(false);
+      keepAlive = null;
     }
   }
 
@@ -333,17 +334,17 @@ public void run() {
 
     @Override
     public void onClose() {
-      final String logMessage = "closed";
-      executorService.execute(
-          new Runnable() {
-            @Override
-            public void run() {
-              if (logger.logsDebug()) {
-                logger.debug(logMessage);
-              }
-              onClosed();
-            }
-          });
+      if (logger.logsDebug()) {
+        logger.debug("closed");
+      }
+      if (!isClosed) {
+        executorService.execute(new Runnable() {
+          @Override
+          public void run() {
+            onClosed();
+          }
+        });
+      }
     }
 
     @Override

src/main/java/com/google/firebase/database/core/Context.java

@@ -130,8 +130,12 @@ private void restartServices() {
 
   void stop() {
     stopped = true;
-    eventTarget.shutdown();
-    runLoop.shutdown();
+    if (eventTarget != null) {
+      eventTarget.shutdown();
+    }
+    if (runLoop != null) {
+      runLoop.shutdown();
+    }
   }
 
   protected void assertUnfrozen() {

src/main/java/com/google/firebase/database/core/RepoManager.java

@@ -49,6 +49,14 @@ public static Repo createRepo(Context ctx, RepoInfo info, FirebaseDatabase datab
     return instance.createLocalRepo(ctx, info, database);
   }
 
+  public static void destroy(Context ctx) {
+    try {
+      instance.destroyInternal(ctx);
+    } finally {
+      ctx.stop();
+    }
+  }
+
   public static void interrupt(Context ctx) {
     instance.interruptInternal(ctx);
   }
@@ -134,6 +142,26 @@ public void run() {
     }
   }
 
+  private void destroyInternal(final Context ctx) {
+    RunLoop runLoop = ctx.getRunLoop();
+    if (runLoop != null) {
+      // RunLoop gets initialized before any Repo. Therefore, we can assume that when RunLoop
+      // is not present, there's nothing to cleanup.
+      runLoop.scheduleNow(new Runnable() {
+        @Override
+        public void run() {
+          synchronized (repos) {
+            if (repos.containsKey(ctx)) {
+              for (Repo repo : repos.get(ctx).values()) {
+                repo.interrupt();
+              }
+            }
+          }
+        }
+      });
+    }
+  }
+
   private void resumeInternal(final Context ctx) {
     RunLoop runLoop = ctx.getRunLoop();
     if (runLoop != null) {

src/main/java/com/google/firebase/internal/RevivingScheduledExecutor.java

@@ -91,6 +91,7 @@ public RevivingScheduledExecutor(
     INSTANCE_COUNTER.incrementAndGet();
     this.initialDelayMs = initialDelayMs;
     this.timeoutMs = timeoutMs;
+    setRemoveOnCancelPolicy(true);
     setThreadFactory(
         new ThreadFactory() {
           @Override