DEV: Await for all async MessageBus callbacks (#17966)

Should take care of yet another category of flaky tests

Author: CvX

.eslintrc

@@ -2,7 +2,8 @@
   "extends": "eslint-config-discourse",
   "rules": {
     "discourse-ember/global-ember": 2,
-    "eol-last": 2
+    "eol-last": 2,
+    "no-restricted-globals": 0
   },
   "globals": {
     "_": "off",

app/assets/javascripts/discourse/app/services/presence.js

@@ -1,7 +1,7 @@
 import Service from "@ember/service";
 import EmberObject, { computed } from "@ember/object";
 import { ajax } from "discourse/lib/ajax";
-import { cancel, debounce, next, once, run, throttle } from "@ember/runloop";
+import { cancel, debounce, next, once, throttle } from "@ember/runloop";
 import discourseLater from "discourse-common/lib/later";
 import Session from "discourse/models/session";
 import { Promise } from "rsvp";
@@ -170,15 +170,13 @@ class PresenceChannelState extends EmberObject.extend(Evented) {
 
     this.lastSeenId = initialData.last_message_id;
 
-    let callback = (data, global_id, message_id) =>
-      run(() => this._processMessage(data, global_id, message_id));
     this.presenceService.messageBus.subscribe(
       `/presence${this.name}`,
-      callback,
+      this._processMessage,
       this.lastSeenId
     );
 
-    this.set("_subscribedCallback", callback);
+    this.set("_subscribedCallback", this._processMessage);
     this.trigger("change");
   }
 
@@ -198,12 +196,10 @@ class PresenceChannelState extends EmberObject.extend(Evented) {
 
   async _resubscribe() {
     this.unsubscribe();
-    // Stored at object level for tests to hook in
-    this._resubscribePromise = this.subscribe();
-    await this._resubscribePromise;
-    delete this._resubscribePromise;
+    await this.subscribe();
   }
 
+  @bind
   async _processMessage(data, global_id, message_id) {
     if (message_id !== this.lastSeenId + 1) {
       // eslint-disable-next-line no-console

app/assets/javascripts/discourse/tests/helpers/qunit-helpers.js

@@ -15,7 +15,6 @@ import { getApplication, getContext, settled } from "@ember/test-helpers";
 import { getOwner } from "discourse-common/lib/get-owner";
 import { run } from "@ember/runloop";
 import { setupApplicationTest } from "ember-qunit";
-import { Promise } from "rsvp";
 import Site from "discourse/models/site";
 import User from "discourse/models/user";
 import { _clearSnapshots } from "select-kit/components/composer-actions";
@@ -448,15 +447,11 @@ QUnit.assert.containsInstance = function (collection, klass, message) {
 };
 
 export async function selectDate(selector, date) {
-  return new Promise((resolve) => {
-    const elem = document.querySelector(selector);
-    elem.value = date;
-    const evt = new Event("input", { bubbles: true, cancelable: false });
-    elem.dispatchEvent(evt);
-    elem.blur();
-
-    resolve();
-  });
+  const elem = document.querySelector(selector);
+  elem.value = date;
+  const evt = new Event("input", { bubbles: true, cancelable: false });
+  elem.dispatchEvent(evt);
+  elem.blur();
 }
 
 export function queryAll(selector, context) {
@@ -491,10 +486,12 @@ export function exists(selector) {
 
 export async function publishToMessageBus(channelPath, ...args) {
   args = cloneJSON(args);
-  MessageBus.callbacks
+
+  const promises = MessageBus.callbacks
     .filterBy("channel", channelPath)
-    .forEach((c) => c.func(...args));
+    .map((callback) => callback.func(...args));
 
+  await Promise.allSettled(promises);
   await settled();
 }
 

app/assets/javascripts/discourse/tests/unit/services/presence-test.js

@@ -138,8 +138,6 @@ acceptance("Presence - Subscribing", function (needs) {
       99
     );
 
-    await channel._presenceState._resubscribePromise;
-
     sinon.assert.calledOnce(stub);
     assert.strictEqual(
       channel.users.length,
@@ -189,8 +187,6 @@ acceptance("Presence - Subscribing", function (needs) {
       3
     );
 
-    await channel._presenceState._resubscribePromise;
-
     assert.strictEqual(
       channel.count,
       3,