def LeaseClientActionRequests(self,
                                client_id,
                                lease_time=None,
                                limit=None,
                                cursor=None):
    """Leases available client messages for the client with the given id."""

    now = rdfvalue.RDFDatetime.Now()
    now_str = mysql_utils.RDFDatetimeToTimestamp(now)
    expiry = now + lease_time
    expiry_str = mysql_utils.RDFDatetimeToTimestamp(expiry)
    proc_id_str = utils.ProcessIdString()
    client_id_int = db_utils.ClientIDToInt(client_id)

    query = ("UPDATE client_action_requests "
             "SET leased_until=FROM_UNIXTIME(%s), leased_by=%s, "
             "leased_count=leased_count+1 "
             "WHERE client_id=%s AND "
             "(leased_until IS NULL OR leased_until < FROM_UNIXTIME(%s)) "
             "LIMIT %s")
    args = [expiry_str, proc_id_str, client_id_int, now_str, limit]

    num_leased = cursor.execute(query, args)
    if num_leased == 0:
      return []

    query = ("SELECT request, leased_count FROM client_action_requests "
             "WHERE client_id=%s AND leased_until=FROM_UNIXTIME(%s) "
             "AND leased_by=%s")

    cursor.execute(query, [client_id_int, expiry_str, proc_id_str])

    ret = []
    expired = []
    for req, leased_count in cursor.fetchall():
      request = rdf_flows.ClientActionRequest.FromSerializedString(req)
      request.leased_by = proc_id_str
      request.leased_until = expiry
      request.ttl = db.Database.CLIENT_MESSAGES_TTL - leased_count
      # > comparison since this check happens after the lease.
      if leased_count > db.Database.CLIENT_MESSAGES_TTL:
        expired.append((request.client_id, request.flow_id, request.request_id))
      else:
        ret.append(request)

    if expired:
      self._DeleteClientActionRequest(expired, cursor=cursor)

    return sorted(ret, key=lambda req: (req.flow_id, req.request_id))
Leases available client messages for the client with the given id.