Drains the client's Task Scheduler queue.

    1) Get all messages in the client queue.
    2) Sort these into a set of session_ids.
    3) Use data_store.DB.ResolvePrefix() to query all requests.
    4) Delete all responses for retransmitted messages (if needed).

    Args:
       client: The ClientURN object specifying this client.
       max_count: The maximum number of messages we will issue for the client.
         If not given, uses self.max_queue_size .

    Returns:
       The tasks respresenting the messages returned. If we can not send them,
       we can reschedule them for later.

def DrainTaskSchedulerQueueForClient(self, client, max_count=None):
    """Drains the client's Task Scheduler queue.

    1) Get all messages in the client queue.
    2) Sort these into a set of session_ids.
    3) Use data_store.DB.ResolvePrefix() to query all requests.
    4) Delete all responses for retransmitted messages (if needed).

    Args:
       client: The ClientURN object specifying this client.
       max_count: The maximum number of messages we will issue for the client.
         If not given, uses self.max_queue_size .

    Returns:
       The tasks respresenting the messages returned. If we can not send them,
       we can reschedule them for later.
    """
    if max_count is None:
      max_count = self.max_queue_size

    if max_count <= 0:
      return []

    client = rdf_client.ClientURN(client)
    start_time = time.time()
    # Drain the queue for this client
    if data_store.RelationalDBEnabled():
      action_requests = data_store.REL_DB.LeaseClientActionRequests(
          client.Basename(),
          lease_time=rdfvalue.Duration.FromSeconds(self.message_expiry_time),
          limit=max_count)
      result = [
          rdf_flow_objects.GRRMessageFromClientActionRequest(r)
          for r in action_requests
      ]
    else:
      new_tasks = queue_manager.QueueManager(token=self.token).QueryAndOwn(
          queue=client.Queue(),
          limit=max_count,
          lease_seconds=self.message_expiry_time)

      initial_ttl = rdf_flows.GrrMessage().task_ttl
      check_before_sending = []
      result = []
      for task in new_tasks:
        if task.task_ttl < initial_ttl - 1:
          # This message has been leased before.
          check_before_sending.append(task)
        else:
          result.append(task)

      if check_before_sending:
        with queue_manager.QueueManager(token=self.token) as manager:
          status_found = manager.MultiCheckStatus(check_before_sending)

          # All messages that don't have a status yet should be sent again.
          for task in check_before_sending:
            if task not in status_found:
              result.append(task)
            else:
              manager.DeQueueClientRequest(task)

    stats_collector_instance.Get().IncrementCounter("grr_messages_sent",
                                                    len(result))
    if result:
      logging.debug("Drained %d messages for %s in %s seconds.", len(result),
                    client,
                    time.time() - start_time)

    return result