Fix flipper connect / disconnect cycle

Summary: Previously we were cancelling the entire context which meant after reconnect nothing was sent. Additionally we now close / reinitiaze the channel so that any old events are not sent on reconnect

Reviewed By: lblasa

Differential Revision: D39658945

fbshipit-source-id: bb02724434aa820d811b49ab799a4643ab7e785a

android/src/main/java/com/facebook/flipper/plugins/uidebugger/UIDebuggerFlipperPlugin.kt

@@ -46,6 +46,7 @@ class UIDebuggerFlipperPlugin(
 
   @Throws(Exception::class)
   override fun onConnect(connection: FlipperConnection) {
+    Log.i(LogTag, "Connected")
     this.context.connectionRef.connection = connection
 
     val rootDescriptor =
@@ -57,15 +58,14 @@ class UIDebuggerFlipperPlugin(
             InitEvent.serializer(), InitEvent(rootDescriptor.getId(context.applicationRef))))
 
     context.treeObserverManager.start()
-    //    nativeScanScheduler.start()
   }
 
   @Throws(Exception::class)
   override fun onDisconnect() {
     this.context.connectionRef.connection = null
-    Log.e(LogTag, "disconnected")
+    Log.i(LogTag, "Disconnected")
 
-    this.nativeScanScheduler.stop()
+    context.treeObserverManager.stop()
   }
 
   override fun runInBackground(): Boolean {

android/src/main/java/com/facebook/flipper/plugins/uidebugger/observers/TreeObserverManager.kt

@@ -29,7 +29,8 @@ data class SubtreeUpdate(
 class TreeObserverManager(val context: Context) {
 
   private val rootObserver = ApplicationTreeObserver(context)
-  private val treeUpdates = Channel<SubtreeUpdate>(Channel.UNLIMITED)
+  private lateinit var treeUpdates: Channel<SubtreeUpdate>
+  private var job: Job? = null
   private val workerScope = CoroutineScope(Dispatchers.IO)
   private val txId = AtomicInteger()
 
@@ -43,52 +44,56 @@ class TreeObserverManager(val context: Context) {
    */
   fun start() {
 
+    treeUpdates = Channel(Channel.UNLIMITED)
     rootObserver.subscribe(context.applicationRef)
 
-    workerScope.launch {
-      while (isActive) {
-        try {
+    job =
+        workerScope.launch {
+          while (isActive) {
+            try {
 
-          val treeUpdate = treeUpdates.receive()
+              val treeUpdate = treeUpdates.receive()
 
-          val onWorkerThread = System.currentTimeMillis()
+              val onWorkerThread = System.currentTimeMillis()
 
-          val txId = txId.getAndIncrement().toLong()
-          val serialized =
-              Json.encodeToString(
-                  SubtreeUpdateEvent.serializer(),
-                  SubtreeUpdateEvent(txId, treeUpdate.observerType, treeUpdate.nodes))
+              val txId = txId.getAndIncrement().toLong()
+              val serialized =
+                  Json.encodeToString(
+                      SubtreeUpdateEvent.serializer(),
+                      SubtreeUpdateEvent(txId, treeUpdate.observerType, treeUpdate.nodes))
 
-          val serializationEnd = System.currentTimeMillis()
+              val serializationEnd = System.currentTimeMillis()
 
-          context.connectionRef.connection?.send(SubtreeUpdateEvent.name, serialized)
-          val socketEnd = System.currentTimeMillis()
-          Log.i(LogTag, "Sent event for ${treeUpdate.observerType} nodes ${treeUpdate.nodes.size}")
+              context.connectionRef.connection?.send(SubtreeUpdateEvent.name, serialized)
+              val socketEnd = System.currentTimeMillis()
+              Log.i(
+                  LogTag,
+                  "Sent event for ${treeUpdate.observerType} nodes ${treeUpdate.nodes.size}")
 
-          val perfStats =
-              PerfStatsEvent(
-                  txId = txId,
-                  observerType = treeUpdate.observerType,
-                  start = treeUpdate.startTime,
-                  traversalComplete = treeUpdate.traversalCompleteTime,
-                  queuingComplete = onWorkerThread,
-                  serializationComplete = serializationEnd,
-                  socketComplete = socketEnd,
-                  nodesCount = treeUpdate.nodes.size)
-          context.connectionRef.connection?.send(
-              PerfStatsEvent.name, Json.encodeToString(PerfStatsEvent.serializer(), perfStats))
-        } catch (e: java.lang.Exception) {
-          Log.e(LogTag, "Error in channel ", e)
+              val perfStats =
+                  PerfStatsEvent(
+                      txId = txId,
+                      observerType = treeUpdate.observerType,
+                      start = treeUpdate.startTime,
+                      traversalComplete = treeUpdate.traversalCompleteTime,
+                      queuingComplete = onWorkerThread,
+                      serializationComplete = serializationEnd,
+                      socketComplete = socketEnd,
+                      nodesCount = treeUpdate.nodes.size)
+              context.connectionRef.connection?.send(
+                  PerfStatsEvent.name, Json.encodeToString(PerfStatsEvent.serializer(), perfStats))
+            } catch (e: CancellationException) {} catch (e: java.lang.Exception) {
+              Log.e(LogTag, "Unexpected Error in channel ", e)
+            }
+          }
+
+          Log.i(LogTag, "Shutting down worker")
         }
-      }
-
-      Log.i(LogTag, "shutting down worker")
-    }
   }
 
   fun stop() {
     rootObserver.cleanUpRecursive()
-    treeUpdates.close()
-    workerScope.cancel()
+    job?.cancel()
+    treeUpdates.cancel()
   }
 }