def DrainTaskSchedulerQueueForClient(self, client, max_count=None):
    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()
    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:
          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)
          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