code: def ProcessMessages(self, active_notifications, queue_manager, time_limit=0):
      
      now = time.time()
      processed = 0
      for notification in active_notifications:
        if notification.session_id not in self.queued_flows:
          if time_limit and time.time() - now > time_limit:
            break
  
          processed += 1
          self.queued_flows.Put(notification.session_id, 1)
          self.thread_pool.AddTask(
              target=self._ProcessMessages,
              args=(notification, queue_manager.Copy()),
              name=self.__class__.__name__)
  
      return processed
docs: Processes all the flows in the messages.
  
      Precondition: All tasks come from the same queue.
  
      Note that the server actually completes the requests in the
      flow when receiving the messages from the client. We do not really
      look at the messages here at all any more - we just work from the
      completed messages in the flow RDFValue.
  
      Args:
          active_notifications: The list of notifications.
          queue_manager: QueueManager object used to manage notifications,
            requests and responses.
          time_limit: If set return as soon as possible after this many seconds.
  
      Returns:
          The number of processed flows.