code: def process_messages(self, max_messages=10000):
          
          subscribe_clients = [self.primary_subscribe_client]
          for subscribe_client in subscribe_clients:
              for _ in range(max_messages):
                  message = subscribe_client.get_message()
                  if message is None:
                      # Continue on to the next subscribe client.
                      break
  
                  # Parse the message.
                  channel = message["channel"]
                  data = message["data"]
  
                  # Determine the appropriate message handler.
                  if channel == ray.gcs_utils.XRAY_HEARTBEAT_BATCH_CHANNEL:
                      # Similar functionality as raylet info channel
                      message_handler = self.xray_heartbeat_batch_handler
                  elif channel == ray.gcs_utils.XRAY_DRIVER_CHANNEL:
                      # Handles driver death.
                      message_handler = self.xray_driver_removed_handler
                  else:
                      raise Exception("This code should be unreachable.")
  
                  # Call the handler.
                  message_handler(channel, data)
docs: Process all messages ready in the subscription channels.
  
          This reads messages from the subscription channels and calls the
          appropriate handlers until there are no messages left.
  
          Args:
              max_messages: The maximum number of messages to process before
                  returning.