Class Queue

layers/containers/highcontention/queue.py:77–285

class Queue:
    # Public functions
    def __init__(self, subspace, highContention=True):
        self.subspace = subspace
        self.highContention = highContention

        self._conflictedPop = self.subspace['pop']
        self._conflictedItem = self.subspace['conflict']
        self._queueItem = self.subspace['item']

    @fdb.transactional
    def clear(self, tr):
        """Remove all items from the queue."""
        del tr[self.subspace.range()]

    @fdb.transactional
    def push(self, tr, value):
        """Push a single item onto the queue."""
        index = self._getNextIndex(tr.snapshot, self._queueItem)
        self._pushAt(tr, self._encodeValue(value), index)

    def pop(self, db):
        """Pop the next item from the queue. Cannot be composed with other functions in a single transaction."""

        if self.highContention:
            result = self._popHighContention(db)
        else:
            result = self._popSimple(db)

        if result is None:
            return result

        return self._decodeValue(result)

    @fdb.transactional
    def empty(self, tr):
        """Test whether the queue is empty."""
        return self._getFirstItem(tr) is None

    @fdb.transactional
    def peek(self, tr):
        """Get the value of the next item in the queue without popping it."""
        firstItem = self._getFirstItem(tr)
        if firstItem is None:
            return None
        else:
            return self._decodeValue(firstItem.value)

    # Private functions

    def _conflictedItemKey(self, subKey):
        return self._conflictedItem.pack((subKey,))

    def _randID(self):
        return os.urandom(20)  # this relies on good random data from the OS to avoid collisions

    def _encodeValue(self, value):
        return fdb.tuple.pack((value,))