simplify control flow in _process_message_queue()

Author: dlqqq

jupyter_rtc_core/rooms/yroom.py

@@ -162,18 +162,33 @@ async def _process_message_queue(self) -> None:
             except asyncio.QueueShutDown:
                 break
 
-            # Handle Awareness messages
+            # Determine message type & subtype from header
             message_type = message[0]
-            if message_type == YMessageType.AWARENESS:
+            sync_message_subtype = "*"
+            if message_type == YMessageType.SYNC and len(message) >= 2:
+                sync_message_subtype = message[1]
+            
+            # Determine if message is invalid
+            invalid_message_type = message_type not in YMessageType
+            invalid_sync_message_type = message_type == YMessageType.SYNC and sync_message_subtype not in YSyncMessageSubtype
+            invalid_message = invalid_message_type or invalid_sync_message_type
+
+            # Handle invalid messages by logging a warning and ignoring
+            if invalid_message:
+                self.log.warning(
+                    "Ignoring an unrecognized message with header "
+                    f"'{message_type},{sync_message_subtype}' from client "
+                    "'{client_id}'. Messages must have one of the following "
+                    "headers: '0,0' (SyncStep1), '0,1' (SyncStep2), "
+                    "'0,2' (SyncUpdate), or '1,*' (AwarenessUpdate)."
+                )
+            # Handle Awareness messages
+            elif message_type == YMessageType.AWARENESS:
                 self.log.debug(f"Received AwarenessUpdate from '{client_id}'.")
                 self.handle_awareness_update(client_id, message)
                 self.log.debug(f"Handled AwarenessUpdate from '{client_id}'.")
-            
             # Handle Sync messages
-            sync_message_subtype = None
-            if message_type == YMessageType.SYNC and len(message) >= 2:
-                sync_message_subtype = message[1]
-            if sync_message_subtype == YSyncMessageSubtype.SYNC_STEP1:
+            elif sync_message_subtype == YSyncMessageSubtype.SYNC_STEP1:
                 self.log.info(f"Received SS1 from '{client_id}'.")
                 self.handle_sync_step1(client_id, message)
                 self.log.info(f"Handled SS1 from '{client_id}'.")
@@ -185,20 +200,17 @@ async def _process_message_queue(self) -> None:
                 self.log.info(f"Received SyncUpdate from '{client_id}'.")
                 self.handle_sync_update(client_id, message)
                 self.log.info(f"Handled SyncUpdate from '{client_id}'.")
-            else:
-                self.log.warning(
-                    "Ignoring an unrecognized message with header "
-                    f"'{message_type},{sync_message_subtype}' from client "
-                    "'{client_id}'. Messages must have one of the following "
-                    "headers: '0,0' (SyncStep1), '0,1' (SyncStep2), "
-                    "'0,2' (SyncUpdate), or '1,*' (AwarenessUpdate)."
-                )
 
             # Finally, inform the asyncio Queue that the task was complete
             # This is required for `self._message_queue.join()` to unblock once
             # queue is empty in `self.stop()`.
             self._message_queue.task_done()
 
+        self.log.info(
+            "Stopped `self._process_message_queue()` background task "
+            f"for YRoom '{self.room_id}'."
+        )
+
 
     def handle_sync_step1(self, client_id: str, message: bytes) -> None:
         """