diff --git a/.changeset/tough-rings-bake.md b/.changeset/tough-rings-bake.md
new file mode 100644
index 00000000000..f02fbd66ab5
--- /dev/null
+++ b/.changeset/tough-rings-bake.md
@@ -0,0 +1,6 @@
+---
+'firebase': patch
+'@firebase/messaging': patch
+---
+
+Added support for `onMessage` so the internal callback can work with [Subscriber](https://rxjs.dev/api/index/class/Subscriber)
diff --git a/packages/messaging/src/controllers/window-controller.test.ts b/packages/messaging/src/controllers/window-controller.test.ts
index 53c9d9a01a3..a504e553a14 100644
--- a/packages/messaging/src/controllers/window-controller.test.ts
+++ b/packages/messaging/src/controllers/window-controller.test.ts
@@ -1,3 +1,25 @@
+import '../testing/setup';
+
+import * as tokenManagementModule from '../core/token-management';
+
+import {
+  CONSOLE_CAMPAIGN_ANALYTICS_ENABLED,
+  CONSOLE_CAMPAIGN_ID,
+  CONSOLE_CAMPAIGN_NAME,
+  CONSOLE_CAMPAIGN_TIME,
+  DEFAULT_SW_PATH,
+  DEFAULT_SW_SCOPE,
+  DEFAULT_VAPID_KEY
+} from '../util/constants';
+import { InternalMessage, MessageType } from '../interfaces/internal-message';
+import { SinonFakeTimers, SinonSpy, spy, stub, useFakeTimers } from 'sinon';
+import { Spy, Stub } from '../testing/sinon-types';
+
+import { ErrorCode } from '../util/errors';
+import { FakeServiceWorkerRegistration } from '../testing/fakes/service-worker';
+import { FirebaseAnalyticsInternal } from '@firebase/analytics-interop-types';
+import { FirebaseInternalDependencies } from '../interfaces/internal-dependencies';
+import { WindowController } from './window-controller';
 /**
  * @license
  * Copyright 2017 Google LLC
@@ -15,27 +37,7 @@
  * limitations under the License.
  */
 import { expect } from 'chai';
-import { stub, spy, SinonSpy, useFakeTimers, SinonFakeTimers } from 'sinon';
-
-import { FirebaseAnalyticsInternal } from '@firebase/analytics-interop-types';
-import { WindowController } from './window-controller';
 import { getFakeFirebaseDependencies } from '../testing/fakes/firebase-dependencies';
-import { ErrorCode } from '../util/errors';
-import { FirebaseInternalDependencies } from '../interfaces/internal-dependencies';
-import * as tokenManagementModule from '../core/token-management';
-import {
-  DEFAULT_VAPID_KEY,
-  DEFAULT_SW_SCOPE,
-  DEFAULT_SW_PATH,
-  CONSOLE_CAMPAIGN_ANALYTICS_ENABLED,
-  CONSOLE_CAMPAIGN_ID,
-  CONSOLE_CAMPAIGN_NAME,
-  CONSOLE_CAMPAIGN_TIME
-} from '../util/constants';
-import { Stub, Spy } from '../testing/sinon-types';
-import '../testing/setup';
-import { FakeServiceWorkerRegistration } from '../testing/fakes/service-worker';
-import { MessageType, InternalMessage } from '../interfaces/internal-message';
 
 type MessageEventListener = (event: Event) => Promise<void>;
 
diff --git a/packages/messaging/src/controllers/window-controller.ts b/packages/messaging/src/controllers/window-controller.ts
index 663d8fc0067..1907224c29c 100644
--- a/packages/messaging/src/controllers/window-controller.ts
+++ b/packages/messaging/src/controllers/window-controller.ts
@@ -15,30 +15,31 @@
  * limitations under the License.
  */
 
-import { getToken, deleteToken } from '../core/token-management';
-import { FirebaseInternalDependencies } from '../interfaces/internal-dependencies';
-import { FirebaseMessaging } from '@firebase/messaging-types';
-import { ERROR_FACTORY, ErrorCode } from '../util/errors';
-import { NextFn, Observer, Unsubscribe } from '@firebase/util';
-import { InternalMessage, MessageType } from '../interfaces/internal-message';
 import {
-  CONSOLE_CAMPAIGN_ID,
   CONSOLE_CAMPAIGN_ANALYTICS_ENABLED,
+  CONSOLE_CAMPAIGN_ID,
   CONSOLE_CAMPAIGN_NAME,
   CONSOLE_CAMPAIGN_TIME,
   DEFAULT_SW_PATH,
   DEFAULT_SW_SCOPE,
   DEFAULT_VAPID_KEY
 } from '../util/constants';
-import { FirebaseApp } from '@firebase/app-types';
+import { ERROR_FACTORY, ErrorCode } from '../util/errors';
+import { InternalMessage, MessageType } from '../interfaces/internal-message';
+import { NextFn, Observer, Unsubscribe } from '@firebase/util';
+import { deleteToken, getToken } from '../core/token-management';
+
 import { ConsoleMessageData } from '../interfaces/message-payload';
-import { isConsoleMessage } from '../helpers/is-console-message';
+import { FirebaseApp } from '@firebase/app-types';
+import { FirebaseInternalDependencies } from '../interfaces/internal-dependencies';
+import { FirebaseMessaging } from '@firebase/messaging-types';
 import { FirebaseService } from '@firebase/app-types/private';
+import { isConsoleMessage } from '../helpers/is-console-message';
 
 export class WindowController implements FirebaseMessaging, FirebaseService {
   private vapidKey: string | null = null;
   private swRegistration?: ServiceWorkerRegistration;
-  private onMessageCallback: NextFn<object> | null = null;
+  private onMessageCallback: NextFn<object> | Observer<object> | null = null;
 
   constructor(
     private readonly firebaseDependencies: FirebaseInternalDependencies
@@ -131,12 +132,8 @@ export class WindowController implements FirebaseMessaging, FirebaseService {
    * message.
    * @return The unsubscribe function for the observer.
    */
-  // TODO: Simplify this to only accept a function and not an Observer.
   onMessage(nextOrObserver: NextFn<object> | Observer<object>): Unsubscribe {
-    this.onMessageCallback =
-      typeof nextOrObserver === 'function'
-        ? nextOrObserver
-        : nextOrObserver.next;
+    this.onMessageCallback = nextOrObserver;
 
     return () => {
       this.onMessageCallback = null;
@@ -193,8 +190,13 @@ export class WindowController implements FirebaseMessaging, FirebaseService {
 
     const { type, payload } = (event.data as InternalMessage).firebaseMessaging;
 
+    // onMessageCallback is either a function or observer/subscriber.
     if (this.onMessageCallback && type === MessageType.PUSH_RECEIVED) {
-      this.onMessageCallback(payload);
+      if (typeof this.onMessageCallback === 'function') {
+        this.onMessageCallback(payload);
+      } else {
+        this.onMessageCallback.next(payload);
+      }
     }
 
     const { data } = payload;