Claims records from a queue. See server/aff4_objects/queue.py.

def QueueClaimRecords(self,
                        queue_id,
                        item_rdf_type,
                        limit=10000,
                        timeout="30m",
                        start_time=None,
                        record_filter=lambda x: False,
                        max_filtered=1000):
    """Claims records from a queue. See server/aff4_objects/queue.py."""
    now = rdfvalue.RDFDatetime.Now()
    expiration = rdfvalue.RDFDatetime.Now() + rdfvalue.Duration(timeout)

    after_urn = None
    if start_time:
      after_urn, _, _ = DataStore.CollectionMakeURN(
          queue_id, start_time.AsMicrosecondsSinceEpoch(), 0, subpath="Records")
    results = []

    filtered_count = 0

    for subject, values in DB.ScanAttributes(
        str(queue_id.Add("Records")),
        [DataStore.COLLECTION_ATTRIBUTE, DataStore.QUEUE_LOCK_ATTRIBUTE],
        max_records=4 * limit,
        after_urn=after_urn):
      if DataStore.COLLECTION_ATTRIBUTE not in values:
        # Unlikely case, but could happen if, say, a thread called RefreshClaims
        # so late that another thread already deleted the record. Go ahead and
        # clean this up.
        self.DeleteAttributes(subject, [DataStore.QUEUE_LOCK_ATTRIBUTE])
        continue
      if DataStore.QUEUE_LOCK_ATTRIBUTE in values:
        timestamp = rdfvalue.RDFDatetime.FromMicrosecondsSinceEpoch(
            values[DataStore.QUEUE_LOCK_ATTRIBUTE][1])
        if timestamp > now:
          continue
      rdf_value = item_rdf_type.FromSerializedString(
          values[DataStore.COLLECTION_ATTRIBUTE][1])
      if record_filter(rdf_value):
        filtered_count += 1
        if max_filtered and filtered_count >= max_filtered:
          break
        continue
      results.append(
          Record(
              queue_id=queue_id,
              timestamp=values[DataStore.COLLECTION_ATTRIBUTE][0],
              suffix=int(subject[-6:], 16),
              subpath="Records",
              value=rdf_value))
      self.Set(subject, DataStore.QUEUE_LOCK_ATTRIBUTE, expiration)

      filtered_count = 0
      if len(results) >= limit:
        break

    return results