def process_messages(self, max_messages=10000):
        """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.
        """
        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)
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 : break channel = message [ "channel" ] data = message [ "data" ] if channel == ray . gcs_utils . XRAY_HEARTBEAT_BATCH_CHANNEL : message_handler = self . xray_heartbeat_batch_handler elif channel == ray . gcs_utils . XRAY_DRIVER_CHANNEL : message_handler = self . xray_driver_removed_handler else : raise Exception ( "This code should be unreachable." ) message_handler ( channel , data )
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.