fix(messaging,android): deliver every foreground message to onMessage exactly once

.github/workflows/android_unit_tests.yaml

@@ -43,7 +43,7 @@ jobs:
         # Tasks are listed per package rather than using the root testDebugUnitTest
         # task, which would also build the Dart sources of the aggregate test app.
         # Add a task here when a package gains an android/src/test directory.
-        run: cd tests/android && ./gradlew :firebase_crashlytics:testDebugUnitTest :cloud_functions:testDebugUnitTest
+        run: cd tests/android && ./gradlew :firebase_crashlytics:testDebugUnitTest :cloud_functions:testDebugUnitTest :firebase_messaging:testDebugUnitTest
       - name: 'Upload test reports'
         if: failure()
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a

packages/firebase_messaging/firebase_messaging/android/build.gradle

@@ -59,12 +59,17 @@ android {
     disable 'InvalidPackage'
   }
 
+  testOptions {
+    unitTests.returnDefaultValues = true
+  }
+
   dependencies {
     api firebaseCoreProject
     implementation platform("com.google.firebase:firebase-bom:${getRootProjectExtOrCoreProperty("FirebaseSDKVersion", firebaseCoreProject)}")
     implementation 'com.google.firebase:firebase-messaging'
     implementation 'androidx.localbroadcastmanager:localbroadcastmanager:1.1.0'
     implementation 'androidx.annotation:annotation:1.7.1'
+    testImplementation 'junit:junit:4.13.2'
   }
 }
 

packages/firebase_messaging/firebase_messaging/android/src/main/java/io/flutter/plugins/firebase/messaging/FlutterFirebaseMessagingPlugin.java

@@ -57,6 +57,10 @@ public class FlutterFirebaseMessagingPlugin
 
   private static final String TAG = "FLTFireMsgPlugin";
 
+  // Messages sent to Dart before it has set up its handler on the channel, e.g. the token or a
+  // message received while the engine starts. Flutter only keeps one by default.
+  private static final int CHANNEL_BUFFER_SIZE = 64;
+
   /** Mirrors the de-duplication window the Messaging SDK keeps in FcmLifecycleCallbacks. */
   private static final int RECENTLY_LOGGED_MESSAGE_IDS_MAX_SIZE = 10;
 
@@ -66,9 +70,10 @@ public class FlutterFirebaseMessagingPlugin
   private MethodChannel channel;
   private Activity mainActivity;
 
-  private final LiveData<RemoteMessage> liveDataRemoteMessage =
-      FlutterFirebaseRemoteMessageLiveData.getInstance();
-  private Observer<RemoteMessage> remoteMessageObserver;
+  private final FlutterFirebaseRemoteMessageDispatcher remoteMessageDispatcher =
+      FlutterFirebaseRemoteMessageDispatcher.getInstance();
+  private FlutterFirebaseRemoteMessageDispatcher.Listener remoteMessageListener;
+  private boolean listeningToForegroundMessages = false;
   private final LiveData<String> liveDataToken = FlutterFirebaseTokenLiveData.getInstance();
   private Observer<String> tokenObserver;
   private final LiveData<String> liveDataRegistered =
@@ -84,24 +89,32 @@ public class FlutterFirebaseMessagingPlugin
 
   FlutterFirebasePermissionManager permissionManager;
 
+  // Sends a message to Dart. Messages sent before Dart has set up its handler on the channel wait
+  // in a buffer that only keeps one by default, so it is enlarged with every message rather than
+  // once at registration: engines created with `new FlutterEngine(...)` register plugins before
+  // they run Dart, and Flutter drops messages until then, and a hot restart resets the buffers.
+  private void invokeDartMethod(String method, Object arguments) {
+    channel.resizeChannelBuffer(CHANNEL_BUFFER_SIZE);
+    channel.invokeMethod(method, arguments);
+  }
+
   private void initInstance(BinaryMessenger messenger) {
     String channelName = "plugins.flutter.io/firebase_messaging";
     channel = new MethodChannel(messenger, channelName);
     channel.setMethodCallHandler(this);
     permissionManager = new FlutterFirebasePermissionManager();
 
-    remoteMessageObserver =
+    remoteMessageListener =
         remoteMessage -> {
           Map<String, Object> content =
               FlutterFirebaseMessagingUtils.remoteMessageToMap(remoteMessage);
-          channel.invokeMethod("Messaging#onMessage", content);
+          invokeDartMethod("Messaging#onMessage", content);
         };
-    tokenObserver = token -> channel.invokeMethod("Messaging#onTokenRefresh", token);
-    registeredObserver = fid -> channel.invokeMethod("Messaging#onRegistered", fid);
-    unregisteredObserver = fid -> channel.invokeMethod("Messaging#onUnregistered", fid);
+    tokenObserver = token -> invokeDartMethod("Messaging#onTokenRefresh", token);
+    registeredObserver = fid -> invokeDartMethod("Messaging#onRegistered", fid);
+    unregisteredObserver = fid -> invokeDartMethod("Messaging#onUnregistered", fid);
     // We remove these observers in the onDetachedFromEngine method. Using "observeForever()"
     // allows us to use without a LifecycleOwner.
-    liveDataRemoteMessage.observeForever(remoteMessageObserver);
     liveDataToken.observeForever(tokenObserver);
     liveDataRegistered.observeForever(registeredObserver);
     liveDataUnregistered.observeForever(unregisteredObserver);
@@ -118,7 +131,8 @@ public void onAttachedToEngine(FlutterPluginBinding binding) {
   @Override
   public void onDetachedFromEngine(@NonNull FlutterPluginBinding binding) {
     liveDataToken.removeObserver(tokenObserver);
-    liveDataRemoteMessage.removeObserver(remoteMessageObserver);
+    remoteMessageDispatcher.removeListener(remoteMessageListener);
+    listeningToForegroundMessages = false;
     liveDataRegistered.removeObserver(registeredObserver);
     liveDataUnregistered.removeObserver(unregisteredObserver);
   }
@@ -577,6 +591,21 @@ public void onMethodCall(final MethodCall call, @NonNull final Result result) {
         FlutterFirebaseMessagingBackgroundService.setUserCallbackHandle(userCallbackHandle);
         methodCallTask = Tasks.forResult(null);
         break;
+      // Sent when Dart starts and stops listening to onMessage. Foreground messages are only sent
+      // to the engines that listen: the messages received while none does are kept until one
+      // does, and engines that never listen (the background engine) get none.
+      case "Messaging#startForegroundMessages":
+        if (!listeningToForegroundMessages) {
+          listeningToForegroundMessages = true;
+          remoteMessageDispatcher.addListener(remoteMessageListener);
+        }
+        methodCallTask = Tasks.forResult(null);
+        break;
+      case "Messaging#stopForegroundMessages":
+        listeningToForegroundMessages = false;
+        remoteMessageDispatcher.removeListener(remoteMessageListener);
+        methodCallTask = Tasks.forResult(null);
+        break;
       case "Messaging#getInitialMessage":
         methodCallTask = getInitialMessage();
         break;
@@ -770,7 +799,7 @@ private boolean handleNotificationIntent(@NonNull Intent intent, boolean shouldN
 
     // On a terminated launch, getInitialMessage() owns this message.
     if (shouldNotifyStream) {
-      channel.invokeMethod("Messaging#onMessageOpenedApp", message);
+      invokeDartMethod("Messaging#onMessageOpenedApp", message);
     }
     mainActivity.setIntent(intent);
     return true;

packages/firebase_messaging/firebase_messaging/android/src/main/java/io/flutter/plugins/firebase/messaging/FlutterFirebaseMessagingReceiver.java

@@ -67,7 +67,7 @@ public void onReceive(Context context, Intent intent) {
             //      App in Foreground
             //   ------------------------
             if (FlutterFirebaseMessagingUtils.isApplicationForeground(context)) {
-              FlutterFirebaseRemoteMessageLiveData.getInstance().postRemoteMessage(remoteMessage);
+              FlutterFirebaseRemoteMessageDispatcher.getInstance().dispatch(remoteMessage);
               return;
             }
 
@@ -91,7 +91,10 @@ public void onReceive(Context context, Intent intent) {
                 onBackgroundMessageIntent,
                 remoteMessage.getOriginalPriority() == RemoteMessage.PRIORITY_HIGH);
           } finally {
-            pendingResult.finish();
+            // Null when onReceive is called outside of a broadcast, as the e2e tests do.
+            if (pendingResult != null) {
+              pendingResult.finish();
+            }
           }
         });
   }

packages/firebase_messaging/firebase_messaging/android/src/main/java/io/flutter/plugins/firebase/messaging/FlutterFirebaseRemoteMessageDispatcher.java

@@ -0,0 +1,141 @@
+// Copyright 2026, the Chromium project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+package io.flutter.plugins.firebase.messaging;
+
+import android.os.Handler;
+import android.os.Looper;
+import android.os.SystemClock;
+import android.util.Log;
+import androidx.annotation.NonNull;
+import androidx.annotation.Nullable;
+import androidx.annotation.VisibleForTesting;
+import com.google.firebase.messaging.RemoteMessage;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executor;
+
+/**
+ * Delivers the messages received while the app is in the foreground to the plugin on every engine
+ * whose Dart listens to {@code onMessage}, each message exactly once.
+ *
+ * <p>This replaces a {@code LiveData}, which keeps only the last of the values posted before the
+ * main thread runs, and hands its last value again to every new observer (each new engine).
+ *
+ * <p>Messages received while no engine is listening, for example while the app engine is starting,
+ * are kept for {@link #PENDING_MESSAGE_LIFETIME_MS} and delivered to the first engine that starts
+ * listening.
+ */
+class FlutterFirebaseRemoteMessageDispatcher {
+  interface Listener {
+    void onRemoteMessage(@NonNull RemoteMessage message);
+  }
+
+  interface Clock {
+    long elapsedRealtime();
+  }
+
+  private static final String TAG = "FLTFireMsgDispatcher";
+
+  @VisibleForTesting static final int MAX_PENDING_MESSAGES = 50;
+  @VisibleForTesting static final long PENDING_MESSAGE_LIFETIME_MS = 60_000;
+
+  private static FlutterFirebaseRemoteMessageDispatcher instance;
+
+  @SuppressWarnings("deprecation")
+  static synchronized FlutterFirebaseRemoteMessageDispatcher getInstance() {
+    if (instance == null) {
+      Handler mainHandler = new Handler(Looper.getMainLooper());
+      instance =
+          new FlutterFirebaseRemoteMessageDispatcher(
+              mainHandler::post,
+              SystemClock::elapsedRealtime,
+              // Keeps observers of the deprecated LiveData working.
+              message ->
+                  FlutterFirebaseRemoteMessageLiveData.getInstance().notifyObservers(message));
+    }
+    return instance;
+  }
+
+  private static final class PendingMessage {
+    final RemoteMessage message;
+    final long receivedAt;
+
+    PendingMessage(RemoteMessage message, long receivedAt) {
+      this.message = message;
+      this.receivedAt = receivedAt;
+    }
+  }
+
+  private final Executor mainThread;
+  private final Clock clock;
+  @Nullable private final Listener everyMessageListener;
+  // Only accessed on the main thread.
+  private final List<Listener> listeners = new ArrayList<>();
+  private final ArrayDeque<PendingMessage> pendingMessages = new ArrayDeque<>();
+  // Logged once until messages are delivered again, apps may never listen.
+  private boolean loggedDroppedMessages = false;
+
+  /**
+   * @param everyMessageListener is told about every message as it arrives, whether or not an engine
+   *     listens.
+   */
+  @VisibleForTesting
+  FlutterFirebaseRemoteMessageDispatcher(
+      Executor mainThread, Clock clock, @Nullable Listener everyMessageListener) {
+    this.mainThread = mainThread;
+    this.clock = clock;
+    this.everyMessageListener = everyMessageListener;
+  }
+
+  /** Delivers {@code message} on the main thread. Can be called from any thread. */
+  void dispatch(@NonNull RemoteMessage message) {
+    mainThread.execute(() -> deliver(message));
+  }
+
+  /** Must be called on the main thread. */
+  void addListener(@NonNull Listener listener) {
+    listeners.add(listener);
+    loggedDroppedMessages = false;
+    long now = clock.elapsedRealtime();
+    while (!pendingMessages.isEmpty()) {
+      PendingMessage pending = pendingMessages.poll();
+      if (now - pending.receivedAt <= PENDING_MESSAGE_LIFETIME_MS) {
+        listener.onRemoteMessage(pending.message);
+      } else {
+        logDroppedMessage(pending.message, "no engine listened to onMessage in time");
+      }
+    }
+  }
+
+  /** Must be called on the main thread. */
+  void removeListener(@NonNull Listener listener) {
+    listeners.remove(listener);
+  }
+
+  private void logDroppedMessage(RemoteMessage message, String reason) {
+    if (!loggedDroppedMessages) {
+      loggedDroppedMessages = true;
+      Log.w(TAG, "Dropped foreground message " + message.getMessageId() + ": " + reason + ".");
+    }
+  }
+
+  private void deliver(RemoteMessage message) {
+    if (everyMessageListener != null) {
+      everyMessageListener.onRemoteMessage(message);
+    }
+    if (listeners.isEmpty()) {
+      pendingMessages.add(new PendingMessage(message, clock.elapsedRealtime()));
+      if (pendingMessages.size() > MAX_PENDING_MESSAGES) {
+        logDroppedMessage(pendingMessages.poll().message, "too many waiting for onMessage");
+      }
+      return;
+    }
+    // Copy: a listener may remove itself while handling the message.
+    for (Listener listener : new ArrayList<>(listeners)) {
+      listener.onRemoteMessage(message);
+    }
+  }
+}

packages/firebase_messaging/firebase_messaging/android/src/main/java/io/flutter/plugins/firebase/messaging/FlutterFirebaseRemoteMessageLiveData.java

@@ -7,6 +7,13 @@
 import androidx.lifecycle.LiveData;
 import com.google.firebase.messaging.RemoteMessage;
 
+/**
+ * @deprecated Foreground messages are delivered by {@code FlutterFirebaseRemoteMessageDispatcher},
+ *     which, unlike a {@link LiveData}, does not drop messages posted in quick succession or replay
+ *     the last one to new observers. {@link #postRemoteMessage} forwards to it, and observers of
+ *     this LiveData are still told about every message.
+ */
+@Deprecated
 public class FlutterFirebaseRemoteMessageLiveData extends LiveData<RemoteMessage> {
   private static FlutterFirebaseRemoteMessageLiveData instance;
 
@@ -18,6 +25,11 @@ public static FlutterFirebaseRemoteMessageLiveData getInstance() {
   }
 
   public void postRemoteMessage(RemoteMessage remoteMessage) {
-    postValue(remoteMessage);
+    FlutterFirebaseRemoteMessageDispatcher.getInstance().dispatch(remoteMessage);
+  }
+
+  /** Called by the dispatcher on the main thread for every message. */
+  void notifyObservers(RemoteMessage remoteMessage) {
+    setValue(remoteMessage);
   }
 }