Using the SDK ThreadManager for DB websocket threads

Author: hiranya911

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

@@ -3,7 +3,8 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
 
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.google.firebase.internal.GaeThreadFactory;
+import com.google.firebase.internal.RevivingScheduledExecutor;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandler;
@@ -32,6 +33,7 @@
 import io.netty.util.CharsetUtil;
 
 import java.net.URI;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadFactory;
 import javax.net.ssl.SSLException;
 
@@ -40,12 +42,13 @@ class NettyWebSocketClient implements WebsocketConnection.WSClient {
   private final URI uri;
   private final SslContext sslContext;
   private final ChannelHandler channelHandler;
+  private final ExecutorService executorService;
   private final EventLoopGroup group;
 
   private Channel channel;
 
   NettyWebSocketClient(
-      URI uri, String userAgent,
+      URI uri, String userAgent, ThreadFactory threadFactory,
       WebsocketConnection.WSClientEventHandler eventHandler) throws SSLException {
     this.uri = checkNotNull(uri);
     this.sslContext = SslContextBuilder.forClient()
@@ -56,11 +59,9 @@ class NettyWebSocketClient implements WebsocketConnection.WSClient {
         new DefaultHttpHeaders().add("User-Agent", userAgent));
     this.channelHandler = new WebSocketClientHandler(eventHandler, handshaker);
 
-    ThreadFactory factory = new ThreadFactoryBuilder()
-        .setNameFormat("hkj-websocket-%d")
-        .setDaemon(true)
-        .build();
-    this.group = new NioEventLoopGroup(1, factory);
+    this.executorService = new RevivingScheduledExecutor(
+        threadFactory, "firebase-websocket-worker", GaeThreadFactory.isAvailable());
+    this.group = new NioEventLoopGroup(1, this.executorService);
   }
 
   @Override
@@ -91,6 +92,7 @@ public void close() {
       channel.close();
     } finally {
       group.shutdownGracefully();
+      executorService.shutdown();
       // TODO(hkj): https://github.com/netty/netty/issues/7310
     }
   }

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

@@ -227,14 +227,18 @@ public void merge(List<String> path, Map<String, Object> data, RequestResultCall
 
   @Override
   public void purgeOutstandingWrites() {
+    purgeOutstandingWrites("write_cancelled");
+  }
+
+  private void purgeOutstandingWrites(String reason) {
     for (OutstandingPut put : this.outstandingPuts.values()) {
       if (put.onComplete != null) {
-        put.onComplete.onRequestResult("write_canceled", null);
+        put.onComplete.onRequestResult(reason, null);
       }
     }
     for (OutstandingDisconnect onDisconnect : this.onDisconnectRequestQueue) {
       if (onDisconnect.onComplete != null) {
-        onDisconnect.onComplete.onRequestResult("write_canceled", null);
+        onDisconnect.onComplete.onRequestResult(reason, null);
       }
     }
     this.outstandingPuts.clear();
@@ -311,13 +315,14 @@ public void onDisconnect(Connection.DisconnectReason reason) {
 
   @Override
   public void onKill(String reason) {
-    if (logger.logsDebug()) {
-      logger.debug(
-          "Firebase Database connection was forcefully killed by the server. Will not attempt "
-              + "reconnect. Reason: "
-              + reason);
-    }
+    // This usually represents a configuration error by the user (e.g. incorrect database URL).
+    // Log the details as a warning, and purge outstanding writes so the error details bubble
+    // up as exceptions.
+    logger.warn(
+        "Firebase Database connection was forcefully killed by the server. Will not attempt "
+        + "reconnect. Reason: " + reason);
     interrupt(SERVER_KILL_INTERRUPT_REASON);
+    purgeOutstandingWrites("disconnected");
   }
 
   @Override

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

@@ -72,7 +72,8 @@ private WSClient createConnection(
         host, hostInfo.isSecure(), hostInfo.getNamespace(), optLastSessionId);
     try {
       return new NettyWebSocketClient(
-          uri, connectionContext.getUserAgent(), new WSClientHandlerImpl());
+          uri, connectionContext.getUserAgent(), connectionContext.getThreadFactory(),
+          new WSClientHandlerImpl());
     } catch (Exception e) {
       String msg = "Error while initializing websocket client";
       logger.error(msg, e);

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

@@ -143,12 +143,22 @@ public void run() {
   }
 
   private void destroyInternal(final Context ctx) {
-    synchronized (repos) {
-      if (repos.containsKey(ctx)) {
-        for (Repo repo : repos.get(ctx).values()) {
-          repo.interrupt();
+    // RunLoop gets initialized before any Repo is created. Therefore we can assume that when
+    // the RunLoop is not present, there's nothing to clean up.
+    RunLoop runLoop = ctx.getRunLoop();
+    if (runLoop != null) {
+      runLoop.scheduleNow(new Runnable() {
+        @Override
+        public void run() {
+          synchronized (repos) {
+            if (repos.containsKey(ctx)) {
+              for (Repo repo : repos.get(ctx).values()) {
+                repo.interrupt();
+              }
+            }
+          }
         }
-      }
+      });
     }
   }