Fix race condition so all messages are processed before dispatching next request (#1234)

wwwillchen · web-flow · commit 09adecb85f54 · 2025-03-12T15:12:01.000-07:00
diff --git a/mesop/web/src/services/channel.ts b/mesop/web/src/services/channel.ts
@@ -72,7 +72,12 @@ export class Channel {
   private overridedTitle = '';
 
   private messageQueue: QueuedMessage[] = [];
-  private isProcessingMessage = false;
+
+  private processingMessageDeferred: {
+    promise: Promise<void>;
+    resolve: () => void;
+    reject: (reason?: any) => void;
+  } | null = null;
 
   constructor(
     private title: Title,
@@ -137,14 +142,15 @@ export class Channel {
 
     this.eventSource.addEventListener('message', (e) => {
       // Looks like Angular has a bug where it's not intercepting EventSource onmessage.
-      zone.run(() => {
+      zone.run(async () => {
         const data = (e as any).data;
         if (data === STREAM_END) {
           this.eventSource.close();
           this.status = ChannelStatus.CLOSED;
           clearTimeout(this.isWaitingTimeout);
           this.isWaiting = false;
           this._isHotReloading = false;
+          await this.processMessageQueue();
           this.dequeueEvent();
           return;
         }
@@ -185,7 +191,7 @@ export class Channel {
     };
 
     this.webSocket.onmessage = (event) => {
-      zone.run(() => {
+      zone.run(async () => {
         const prefix = 'data: ';
         const payloadData = (
           event.data.slice(prefix.length) as string
@@ -194,6 +200,7 @@ export class Channel {
         if (payloadData === STREAM_END) {
           this._isHotReloading = false;
           this.status = ChannelStatus.CLOSED;
+          await this.processMessageQueue();
           this.dequeueEvent();
           return;
         }
@@ -357,28 +364,32 @@ export class Channel {
       request,
       response,
     });
-    this.processNextMessage();
+    this.processMessageQueue();
   }
 
-  private async processNextMessage() {
-    if (this.isProcessingMessage || this.messageQueue.length === 0) {
-      return;
+  private async processMessageQueue() {
+    if (this.processingMessageDeferred) {
+      return this.processingMessageDeferred.promise;
     }
 
-    this.isProcessingMessage = true;
-    try {
+    this.processingMessageDeferred = createDeferred();
+
+    while (this.messageQueue.length > 0) {
       const queuedMessage = this.messageQueue.shift()!;
-      await this.handleUiResponse(
-        queuedMessage.request,
-        queuedMessage.response,
-        this.initParams,
-      );
-    } finally {
-      this.isProcessingMessage = false;
-      if (this.messageQueue.length > 0) {
-        await this.processNextMessage();
+      try {
+        await this.handleUiResponse(
+          queuedMessage.request,
+          queuedMessage.response,
+          this.initParams,
+        );
+      } catch (error) {
+        console.error('Error handling UI response:', error);
       }
     }
+
+    // All queued messages processed; resolve the promise and clear it.
+    this.processingMessageDeferred.resolve();
+    this.processingMessageDeferred = null;
   }
 
   dispatch(userEvent: UserEvent) {
@@ -506,6 +517,20 @@ export class Channel {
   }
 }
 
+function createDeferred<T = void>(): {
+  promise: Promise<T>;
+  resolve: (value: T | PromiseLike<T>) => void;
+  reject: (reason?: any) => void;
+} {
+  let resolve: (value: T | PromiseLike<T>) => void = () => {};
+  let reject: (reason?: any) => void = () => {};
+  const promise = new Promise<T>((res, rej) => {
+    resolve = res;
+    reject = rej;
+  });
+  return {promise, resolve, reject};
+}
+
 function generatePayloadString(request: UiRequest): string {
   request.setPath(window.location.pathname);
   const array = request.serializeBinary();
