Fix race condition introduced by background platform channels

shell/platform/android/io/flutter/embedding/engine/dart/DartExecutor.java

@@ -286,6 +286,12 @@ public void notifyLowMemoryWarning() {
     }
   }
 
+  @Override
+  public void enableBufferingIncomingMessages() {}
+
+  @Override
+  public void disableBufferingIncomingMessages() {}
+
   /**
    * Configuration options that specify which Dart entrypoint function is executed and where to find
    * that entrypoint and other assets required for Dart execution.
@@ -461,5 +467,11 @@ public void setMessageHandler(
         @Nullable TaskQueue taskQueue) {
       messenger.setMessageHandler(channel, handler, taskQueue);
     }
+
+    @Override
+    public void enableBufferingIncomingMessages() {}
+
+    @Override
+    public void disableBufferingIncomingMessages() {}
   }
 }

shell/platform/android/io/flutter/embedding/engine/dart/DartMessenger.java

@@ -14,9 +14,10 @@
 import io.flutter.plugin.common.BinaryMessenger;
 import java.nio.ByteBuffer;
 import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
 import java.util.Map;
 import java.util.WeakHashMap;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -33,22 +34,37 @@ class DartMessenger implements BinaryMessenger, PlatformMessageHandler {
 
   @NonNull private final FlutterJNI flutterJNI;
 
-  @NonNull private final ConcurrentHashMap<String, HandlerInfo> messageHandlers;
+  /**
+   * Maps a channel name to an object that contains the task queue and the handler associated with
+   * the channel.
+   *
+   * <p>Reads and writes to this map must lock {@code handlersLock}.
+   */
+  @NonNull private final Map<String, HandlerInfo> messageHandlers = new HashMap<>();
+
+  /**
+   * Maps a channel name to an object that holds information about the incoming Dart message.
+   *
+   * <p>Reads and writes to this map must lock {@code handlersLock}.
+   */
+  @NonNull private Map<String, List<BufferedMessageInfo>> bufferedMessages = new HashMap<>();
 
-  @NonNull private final Map<Integer, BinaryMessenger.BinaryReply> pendingReplies;
+  @NonNull private final Object handlersLock = new Object();
+  @NonNull private final AtomicBoolean enableBufferingIncomingMessages = new AtomicBoolean(false);
+
+  @NonNull private final Map<Integer, BinaryMessenger.BinaryReply> pendingReplies = new HashMap<>();
   private int nextReplyId = 1;
 
   @NonNull private final DartMessengerTaskQueue platformTaskQueue = new PlatformTaskQueue();
 
-  @NonNull private WeakHashMap<TaskQueue, DartMessengerTaskQueue> createdTaskQueues;
+  @NonNull
+  private WeakHashMap<TaskQueue, DartMessengerTaskQueue> createdTaskQueues =
+      new WeakHashMap<TaskQueue, DartMessengerTaskQueue>();
 
   @NonNull private TaskQueueFactory taskQueueFactory;
 
   DartMessenger(@NonNull FlutterJNI flutterJNI, @NonNull TaskQueueFactory taskQueueFactory) {
     this.flutterJNI = flutterJNI;
-    this.messageHandlers = new ConcurrentHashMap<>();
-    this.pendingReplies = new HashMap<>();
-    this.createdTaskQueues = new WeakHashMap<TaskQueue, DartMessengerTaskQueue>();
     this.taskQueueFactory = taskQueueFactory;
   }
 
@@ -78,6 +94,10 @@ public DartMessengerTaskQueue makeBackgroundTaskQueue() {
     }
   }
 
+  /**
+   * Holds information about a platform handler, such as the task queue that processes messages from
+   * Dart.
+   */
   private static class HandlerInfo {
     @NonNull public final BinaryMessenger.BinaryMessageHandler handler;
     @Nullable public final DartMessengerTaskQueue taskQueue;
@@ -90,7 +110,22 @@ private static class HandlerInfo {
     }
   }
 
-  /** A serial task queue that can run on a concurrent ExecutorService. */
+  /**
+   * Holds information that allows to dispatch a Dart message to a platform handler when it becomes
+   * available.
+   */
+  private static class BufferedMessageInfo {
+    @NonNull public final ByteBuffer message;
+    int replyId;
+    long messageData;
+
+    BufferedMessageInfo(@NonNull ByteBuffer message, int replyId, long messageData) {
+      this.message = message;
+      this.replyId = replyId;
+      this.messageData = messageData;
+    }
+  }
+
   static class DefaultTaskQueue implements DartMessengerTaskQueue {
     @NonNull private final ExecutorService executor;
     @NonNull private final ConcurrentLinkedQueue<Runnable> queue;
@@ -154,18 +189,53 @@ public void setMessageHandler(
       @Nullable TaskQueue taskQueue) {
     if (handler == null) {
       Log.v(TAG, "Removing handler for channel '" + channel + "'");
-      messageHandlers.remove(channel);
-    } else {
-      DartMessengerTaskQueue dartMessengerTaskQueue = null;
-      if (taskQueue != null) {
-        dartMessengerTaskQueue = createdTaskQueues.get(taskQueue);
-        if (dartMessengerTaskQueue == null) {
-          throw new IllegalArgumentException(
-              "Unrecognized TaskQueue, use BinaryMessenger to create your TaskQueue (ex makeBackgroundTaskQueue).");
-        }
+      synchronized (handlersLock) {
+        messageHandlers.remove(channel);
       }
-      Log.v(TAG, "Setting handler for channel '" + channel + "'");
+      return;
+    }
+    DartMessengerTaskQueue dartMessengerTaskQueue = null;
+    if (taskQueue != null) {
+      dartMessengerTaskQueue = createdTaskQueues.get(taskQueue);
+      if (dartMessengerTaskQueue == null) {
+        throw new IllegalArgumentException(
+            "Unrecognized TaskQueue, use BinaryMessenger to create your TaskQueue (ex makeBackgroundTaskQueue).");
+      }
+    }
+    Log.v(TAG, "Setting handler for channel '" + channel + "'");
+
+    List<BufferedMessageInfo> list;
+    synchronized (handlersLock) {
       messageHandlers.put(channel, new HandlerInfo(handler, dartMessengerTaskQueue));
+      list = bufferedMessages.remove(channel);
+      if (list == null) {
+        return;
+      }
+    }
+    for (BufferedMessageInfo info : list) {
+      dispatchMessageToQueue(
+          channel, messageHandlers.get(channel), info.message, info.replyId, info.messageData);
+    }
+  }
+
+  @Override
+  public void enableBufferingIncomingMessages() {
+    enableBufferingIncomingMessages.set(true);
+  }
+
+  @Override
+  public void disableBufferingIncomingMessages() {
+    Map<String, List<BufferedMessageInfo>> pendingMessages;
+    synchronized (handlersLock) {
+      enableBufferingIncomingMessages.set(false);
+      pendingMessages = bufferedMessages;
+      bufferedMessages = new HashMap<>();
+    }
+    for (Map.Entry<String, List<BufferedMessageInfo>> channel : pendingMessages.entrySet()) {
+      for (BufferedMessageInfo info : channel.getValue()) {
+        dispatchMessageToQueue(
+            channel.getKey(), null, info.message, info.replyId, info.messageData);
+      }
     }
   }
 
@@ -218,25 +288,21 @@ private void invokeHandler(
     }
   }
 
-  @Override
-  public void handleMessageFromDart(
-      @NonNull final String channel,
+  private void dispatchMessageToQueue(
+      @NonNull String channel,
+      @Nullable HandlerInfo handlerInfo,
       @Nullable ByteBuffer message,
-      final int replyId,
+      int replyId,
       long messageData) {
-    // Called from the ui thread.
-    Log.v(TAG, "Received message from Dart over channel '" + channel + "'");
-    @Nullable final HandlerInfo handlerInfo = messageHandlers.get(channel);
-    @Nullable
     final DartMessengerTaskQueue taskQueue = (handlerInfo != null) ? handlerInfo.taskQueue : null;
     Runnable myRunnable =
         () -> {
           Trace.beginSection("DartMessenger#handleMessageFromDart on " + channel);
           try {
             invokeHandler(handlerInfo, message, replyId);
             if (message != null && message.isDirect()) {
-              // This ensures that if a user retains an instance to the ByteBuffer and it happens to
-              // be direct they will get a deterministic error.
+              // This ensures that if a user retains an instance to the ByteBuffer and it
+              // happens to be direct they will get a deterministic error.
               message.limit(0);
             }
           } finally {
@@ -245,12 +311,43 @@ public void handleMessageFromDart(
             Trace.endSection();
           }
         };
-    @NonNull
     final DartMessengerTaskQueue nonnullTaskQueue =
         taskQueue == null ? platformTaskQueue : taskQueue;
     nonnullTaskQueue.dispatch(myRunnable);
   }
 
+  @Override
+  public void handleMessageFromDart(
+      @NonNull String channel, @Nullable ByteBuffer message, int replyId, long messageData) {
+    // Called from the ui thread.
+    Log.v(TAG, "Received message from Dart over channel '" + channel + "'");
+
+    HandlerInfo handlerInfo;
+    boolean messageDeferred;
+    synchronized (handlersLock) {
+      handlerInfo = messageHandlers.get(channel);
+      messageDeferred = (enableBufferingIncomingMessages.get() && handlerInfo == null);
+      if (messageDeferred) {
+        // The channel is not defined when the Dart VM sends a message before the channels are
+        // registered.
+        //
+        // This is possible if the Dart VM starts before channel registration, and if the thread
+        // that registers the channels is busy or slow at registering the channel handlers.
+        //
+        // In such cases, the task dispatchers are queued, and processed when the channel is
+        // defined.
+        if (!bufferedMessages.containsKey(channel)) {
+          bufferedMessages.put(channel, new LinkedList<>());
+        }
+        List<BufferedMessageInfo> buffer = bufferedMessages.get(channel);
+        buffer.add(new BufferedMessageInfo(message, replyId, messageData));
+      }
+    }
+    if (!messageDeferred) {
+      dispatchMessageToQueue(channel, handlerInfo, message, replyId, messageData);
+    }
+  }
+
   @Override
   public void handlePlatformMessageResponse(int replyId, @Nullable ByteBuffer reply) {
     Log.v(TAG, "Received message reply from Dart.");

shell/platform/android/io/flutter/plugin/common/BinaryMessenger.java

@@ -116,6 +116,22 @@ default void setMessageHandler(
     setMessageHandler(channel, handler);
   }
 
+  /**
+   * Enables the ability to queue messages received from Dart.
+   *
+   * <p>This is useful when there are pending channel handler registrations. For example, Dart may
+   * be initialized concurrently, and prior to the registration of the channel handlers. This
+   * implies that Dart may start sending messages while plugins are being registered.
+   */
+  void enableBufferingIncomingMessages();
+
+  /**
+   * Disables the ability to queue messages received from Dart.
+   *
+   * <p>This can be used after all pending channel handlers have been registered.
+   */
+  void disableBufferingIncomingMessages();
+
   /** Handler for incoming binary messages from Flutter. */
   interface BinaryMessageHandler {
     /**

shell/platform/android/io/flutter/view/FlutterNativeView.java

@@ -159,6 +159,12 @@ public void setMessageHandler(String channel, BinaryMessageHandler handler, Task
     dartExecutor.getBinaryMessenger().setMessageHandler(channel, handler, taskQueue);
   }
 
+  @Override
+  public void enableBufferingIncomingMessages() {}
+
+  @Override
+  public void disableBufferingIncomingMessages() {}
+
   /*package*/ FlutterJNI getFlutterJNI() {
     return mFlutterJNI;
   }

shell/platform/android/io/flutter/view/FlutterView.java

@@ -347,6 +347,12 @@ public void removeFirstFrameListener(FirstFrameListener listener) {
     mFirstFrameListeners.remove(listener);
   }
 
+  @Override
+  public void enableBufferingIncomingMessages() {}
+
+  @Override
+  public void disableBufferingIncomingMessages() {}
+
   /**
    * Reverts this back to the {@link SurfaceView} defaults, at the back of its window and opaque.
    */