def test_scope_queue_name(self):
        self.assertEqual(utils.scope_queue_name('my-q'), '/my-q')
        self.assertEqual(utils.scope_queue_name('my-q', None), '/my-q')
        self.assertEqual(utils.scope_queue_name('my-q', '123'), '123/my-q')

        self.assertEqual(utils.scope_queue_name(None), '/')
        self.assertEqual(utils.scope_queue_name(None, '123'), '123/')

    def _count(self, queue_name, project=None, include_claimed=False):
        """Return total number of messages in a queue.

        This method is designed to very quickly count the number
        of messages in a given queue. Expired messages are not
        counted, of course. If the queue does not exist, the
        count will always be 0.

        Note: Some expired messages may be included in the count if
            they haven't been GC'd yet. This is done for performance.
        """
        query = {
            # Messages must belong to this queue and project.
            PROJ_QUEUE: utils.scope_queue_name(queue_name, project),

            # NOTE(kgriffs): Messages must be finalized (i.e., must not
            # be part of an unfinalized transaction).
            #
            # See also the note wrt 'tx' within the definition
            # of ACTIVE_INDEX_FIELDS.
            'tx': None,
        }

        if not include_claimed:
            # Exclude messages that are claimed
            query['c.e'] = {'$lte': timeutils.utcnow_ts()}

        collection = self._collection(queue_name, project)
        return collection.count(filter=query, hint=COUNTING_INDEX_FIELDS)

    def _create(self, name, metadata=None, project=None):
        # NOTE(flaper87): If the connection fails after it was called
        # and we retry to insert the queue, we could end up returning
        # `False` because of the `DuplicatedKeyError` although the
        # queue was indeed created by this API call.
        #
        # TODO(kgriffs): Commented out `retries_on_autoreconnect` for
        # now due to the above issue, since creating a queue is less
        # important to make super HA.

        try:
            # NOTE(kgriffs): Start counting at 1, and assume the first
            # message ever posted will succeed and set t to a UNIX
            # "modified at" timestamp.
            counter = {'v': 1, 't': 0}

            scoped_name = utils.scope_queue_name(name, project)
            self._collection.insert_one(
                {'p_q': scoped_name, 'm': metadata or {},
                 'c': counter})

        except pymongo.errors.DuplicateKeyError:
            return False
        else:
            return True

    def _claimed(self, queue_name, claim_id,
                 expires=None, limit=None, project=None):

        if claim_id is None:
            claim_id = {'$ne': None}

        query = {
            PROJ_QUEUE: utils.scope_queue_name(queue_name, project),
            'c.id': claim_id,
            'c.e': {'$gt': expires or timeutils.utcnow_ts()},
        }

        # NOTE(kgriffs): Claimed messages bust be queried from
        # the primary to avoid a race condition caused by the
        # multi-phased "create claim" algorithm.
        preference = pymongo.read_preferences.ReadPreference.PRIMARY
        collection = self._collection(queue_name, project)
        msgs = collection.find(query, sort=[('k', 1)],
                               read_preference=preference).hint(
                                   CLAIMED_INDEX_FIELDS)

        if limit is not None:
            msgs = msgs.limit(limit)

        now = timeutils.utcnow_ts()

        def denormalizer(msg):
            doc = _basic_message(msg, now)
            doc['claim'] = msg['c']

            return doc

        return utils.HookedCursor(msgs, denormalizer)

    def bulk_delete(self, queue_name, message_ids, project=None):
        message_ids = [mid for mid in map(utils.to_oid, message_ids) if mid]
        query = {
            '_id': {'$in': message_ids},
            PROJ_QUEUE: utils.scope_queue_name(queue_name, project),
        }

        collection = self._collection(queue_name, project)
        collection.remove(query, w=0)

    def get(self, project, queue):
        fields = {'_id': 0}
        key = utils.scope_queue_name(queue, project)
        entry = self._col.find_one({PRIMARY_KEY: key},
                                   projection=fields)

        if entry is None:
            raise errors.QueueNotMapped(queue, project)

        return _normalize(entry)

    def _purge_queue(self, queue_name, project=None):
        """Removes all messages from the queue.

        Warning: Only use this when deleting the queue; otherwise
        you can cause a side-effect of reseting the marker counter
        which can cause clients to miss tons of messages.

        If the queue does not exist, this method fails silently.

        :param queue_name: name of the queue to purge
        :param project: ID of the project to which the queue belongs
        """
        scope = utils.scope_queue_name(queue_name, project)
        collection = self._collection(queue_name, project)
        collection.delete_many({PROJ_QUEUE: scope})

    def delete(self, queue_name, message_id, project=None, claim=None):
        # NOTE(cpp-cabrera): return early - this is an invalid message
        # id so we won't be able to find it any way
        mid = utils.to_oid(message_id)
        if mid is None:
            return

        collection = self._collection(queue_name, project)

        query = {
            '_id': mid,
            PROJ_QUEUE: utils.scope_queue_name(queue_name, project),
        }

        cid = utils.to_oid(claim)
        if cid is None:
            raise errors.ClaimDoesNotExist(claim, queue_name, project)

        now = timeutils.utcnow_ts()
        cursor = collection.find(query).hint(ID_INDEX_FIELDS)

        try:
            message = next(cursor)
        except StopIteration:
            return

        if claim is None:
            if _is_claimed(message, now):
                raise errors.MessageIsClaimed(message_id)

        else:
            if message['c']['id'] != cid:
                kwargs = {}
                # NOTE(flaper87): In pymongo 3.0 PRIMARY is the default and
                # `read_preference` is read only. We'd need to set it when the
                # client is created.
                # NOTE(kgriffs): Read from primary in case the message
                # was just barely claimed, and claim hasn't made it to
                # the secondary.
                message = collection.find_one(query, **kwargs)

                if message['c']['id'] != cid:
                    if _is_claimed(message, now):
                        raise errors.MessageNotClaimedBy(message_id, claim)

                    raise errors.MessageNotClaimed(message_id)

        collection.delete_one(query)

    def update(self, queue, claim_id, metadata, project=None):
        cid = utils.to_oid(claim_id)
        if cid is None:
            raise errors.ClaimDoesNotExist(claim_id, queue, project)

        now = timeutils.utcnow_ts()
        grace = metadata['grace']
        ttl = metadata['ttl']
        claim_expires = now + ttl
        claim_expires_dt = datetime.datetime.utcfromtimestamp(claim_expires)
        message_ttl = ttl + grace
        message_expires = datetime.datetime.utcfromtimestamp(
            claim_expires + grace)

        msg_ctrl = self.driver.message_controller
        claimed = msg_ctrl._claimed(queue, cid, expires=now,
                                    limit=1, project=project)

        try:
            next(claimed)
        except StopIteration:
            raise errors.ClaimDoesNotExist(claim_id, queue, project)

        meta = {
            'id': cid,
            't': ttl,
            'e': claim_expires,
        }

        # TODO(kgriffs): Create methods for these so we don't interact
        # with the messages collection directly (loose coupling)
        scope = utils.scope_queue_name(queue, project)
        collection = msg_ctrl._collection(queue, project)
        collection.update({'p_q': scope, 'c.id': cid},
                          {'$set': {'c': meta}},
                          upsert=False, multi=True)

        # NOTE(flaper87): Dirty hack!
        # This sets the expiration time to
        # `expires` on messages that would
        # expire before claim.
        collection.update({'p_q': scope,
                           'e': {'$lt': claim_expires_dt},
                           'c.id': cid},
                          {'$set': {'e': message_expires,
                                    't': message_ttl}},
                          upsert=False, multi=True)

    def _unclaim(self, queue_name, claim_id, project=None):
        cid = utils.to_oid(claim_id)

        # NOTE(cpp-cabrera): early abort - avoid a DB query if we're handling
        # an invalid ID
        if cid is None:
            return

        # NOTE(cpp-cabrera):  unclaim by setting the claim ID to None
        # and the claim expiration time to now
        now = timeutils.utcnow_ts()
        scope = utils.scope_queue_name(queue_name, project)
        collection = self._collection(queue_name, project)

        collection.update_many({PROJ_QUEUE: scope, 'c.id': cid},
                               {'$set': {'c': {'id': None, 'e': now}}},
                               upsert=False)

    def delete(self, queue_name, message_id, project=None, claim=None):
        # NOTE(cpp-cabrera): return early - this is an invalid message
        # id so we won't be able to find it any way
        mid = utils.to_oid(message_id)
        if mid is None:
            return

        collection = self._collection(queue_name, project)

        query = {
            '_id': mid,
            PROJ_QUEUE: utils.scope_queue_name(queue_name, project),
        }

        cid = utils.to_oid(claim)
        if cid is None:
            raise errors.ClaimDoesNotExist(queue_name, project, claim)

        now = timeutils.utcnow_ts()
        cursor = collection.find(query).hint(ID_INDEX_FIELDS)

        try:
            message = next(cursor)
        except StopIteration:
            return

        if claim is None:
            if _is_claimed(message, now):
                raise errors.MessageIsClaimed(message_id)

        else:
            if message['c']['id'] != cid:
                # NOTE(kgriffs): Read from primary in case the message
                # was just barely claimed, and claim hasn't made it to
                # the secondary.
                pref = pymongo.read_preferences.ReadPreference.PRIMARY
                message = collection.find_one(query, read_preference=pref)

                if message['c']['id'] != cid:
                    if _is_claimed(message, now):
                        raise errors.MessageNotClaimedBy(message_id, claim)

                    raise errors.MessageNotClaimed(message_id)

        collection.remove(query['_id'], w=0)

    def post(self, queue_name, messages, client_uuid, project=None):
        # NOTE(flaper87): This method should be safe to retry on
        # autoreconnect, since we've a 2-step insert for messages.
        # The worst-case scenario is that we'll increase the counter
        # several times and we'd end up with some non-active messages.

        if not self._queue_ctrl.exists(queue_name, project):
            raise errors.QueueDoesNotExist(queue_name, project)

        # NOTE(flaper87): Make sure the counter exists. This method
        # is an upsert.
        self._get_counter(queue_name, project)
        now = timeutils.utcnow_ts()
        now_dt = datetime.datetime.utcfromtimestamp(now)
        collection = self._collection(queue_name, project)

        messages = list(messages)
        msgs_n = len(messages)
        next_marker = self._inc_counter(queue_name,
                                        project,
                                        amount=msgs_n) - msgs_n

        prepared_messages = []
        for index, message in enumerate(messages):
            msg = {
                PROJ_QUEUE: utils.scope_queue_name(queue_name, project),
                't': message['ttl'],
                'e': now_dt + datetime.timedelta(seconds=message['ttl']),
                'u': client_uuid,
                'c': {'id': None, 'e': now, 'c': 0},
                'd': now + message.get('delay', 0),
                'b': message['body'] if 'body' in message else {},
                'k': next_marker + index,
                'tx': None
                }
            if self.driver.conf.enable_checksum:
                msg['cs'] = s_utils.get_checksum(message.get('body', None))

            prepared_messages.append(msg)

        res = collection.insert_many(prepared_messages,
                                     bypass_document_validation=True)

        return [str(id_) for id_ in res.inserted_ids]

    def bulk_delete(self, queue_name, message_ids, project=None,
                    claim_ids=None):
        message_ids = [mid for mid in map(utils.to_oid, message_ids) if mid]
        if claim_ids:
            claim_ids = [cid for cid in map(utils.to_oid, claim_ids) if cid]
        query = {
            '_id': {'$in': message_ids},
            PROJ_QUEUE: utils.scope_queue_name(queue_name, project),
        }

        collection = self._collection(queue_name, project)
        if claim_ids:
            message_claim_ids = []
            messages = collection.find(query).hint(ID_INDEX_FIELDS)
            for message in messages:
                message_claim_ids.append(message['c']['id'])
            for cid in claim_ids:
                if cid not in message_claim_ids:
                    raise errors.ClaimDoesNotExist(cid, queue_name, project)

        collection.delete_many(query)

 def create(self, queue, subscriber, ttl, options, project=None):
     source = queue
     now = timeutils.utcnow_ts()
     ttl = int(ttl)
     expires = now + ttl
     source_query = {'p_q': utils.scope_queue_name(source, project)}
     target_source = self._queue_collection.find_one(source_query,
                                                     fields={'m': 1,
                                                             '_id': 0})
     if target_source is None:
         raise errors.QueueDoesNotExist(target_source, project)
     try:
         subscription_id = self._collection.insert({'s': source,
                                                    'u': subscriber,
                                                    't': ttl,
                                                    'e': expires,
                                                    'o': options,
                                                    'p': project})
         return subscription_id
     except pymongo.errors.DuplicateKeyError:
         return None

    def get(self, queue_name, message_id, project=None):
        mid = utils.to_oid(message_id)
        if mid is None:
            raise errors.MessageDoesNotExist(message_id, queue_name,
                                             project)

        now = timeutils.utcnow_ts()

        query = {
            '_id': mid,
            PROJ_QUEUE: utils.scope_queue_name(queue_name, project),
        }

        collection = self._collection(queue_name, project)
        message = list(collection.find(query).limit(1).hint(ID_INDEX_FIELDS))

        if not message:
            raise errors.MessageDoesNotExist(message_id, queue_name,
                                             project)

        return _basic_message(message[0], now)

    def pop(self, queue_name, limit, project=None):
        query = {
            PROJ_QUEUE: utils.scope_queue_name(queue_name, project),
        }

        # Only include messages that are not part of
        # any claim, or are part of an expired claim.
        now = timeutils.utcnow_ts()
        query['c.e'] = {'$lte': now}

        collection = self._collection(queue_name, project)
        projection = {'_id': 1, 't': 1, 'b': 1, 'c.id': 1}

        messages = (collection.find_one_and_delete(query,
                                                   projection=projection)
                    for _ in range(limit))

        final_messages = [_basic_message(message, now)
                          for message in messages
                          if message]

        return final_messages

    def bulk_get(self, queue_name, message_ids, project=None):
        message_ids = [mid for mid in map(utils.to_oid, message_ids) if mid]
        if not message_ids:
            return iter([])

        now = timeutils.utcnow_ts()

        # Base query, always check expire time
        query = {
            '_id': {'$in': message_ids},
            PROJ_QUEUE: utils.scope_queue_name(queue_name, project),
        }

        collection = self._collection(queue_name, project)

        # NOTE(flaper87): Should this query
        # be sorted?
        messages = collection.find(query).hint(ID_INDEX_FIELDS)

        def denormalizer(msg):
            return _basic_message(msg, now)

        return utils.HookedCursor(messages, denormalizer)

    def _claimed(self, queue_name, claim_id,
                 expires=None, limit=None, project=None):

        if claim_id is None:
            claim_id = {'$ne': None}

        query = {
            PROJ_QUEUE: utils.scope_queue_name(queue_name, project),
            'c.id': claim_id,
            'c.e': {'$gt': expires or timeutils.utcnow_ts()},
        }

        kwargs = {}
        collection = self._collection(queue_name, project)

        # NOTE(kgriffs): Claimed messages bust be queried from
        # the primary to avoid a race condition caused by the
        # multi-phased "create claim" algorithm.
        # NOTE(flaper87): In pymongo 3.0 PRIMARY is the default and
        # `read_preference` is read only. We'd need to set it when the
        # client is created.
        msgs = collection.find(query, sort=[('k', 1)], **kwargs).hint(
            CLAIMED_INDEX_FIELDS)

        if limit is not None:
            msgs = msgs.limit(limit)

        now = timeutils.utcnow_ts()

        def denormalizer(msg):
            doc = _basic_message(msg, now)
            doc['claim'] = msg['c']

            return doc

        return utils.HookedCursor(msgs, denormalizer)

    def _list(self, queue_name, project=None, marker=None,
              echo=False, client_uuid=None, projection=None,
              include_claimed=False, include_delayed=False,
              sort=1, limit=None):
        """Message document listing helper.

        :param queue_name: Name of the queue to list
        :param project: (Default None) Project `queue_name` belongs to. If
            not specified, queries the "global" namespace/project.
        :param marker: (Default None) Message marker from which to start
            iterating. If not specified, starts with the first message
            available in the queue.
        :param echo: (Default False) Whether to return messages that match
            client_uuid
        :param client_uuid: (Default None) UUID for the client that
            originated this request
        :param projection: (Default None) a list of field names that should be
            returned in the result set or a dict specifying the fields to
            include or exclude
        :param include_claimed: (Default False) Whether to include
            claimed messages, not just active ones
        :param include_delayed: (Default False) Whether to include
            delayed messages, not just active ones
        :param sort: (Default 1) Sort order for the listing. Pass 1 for
            ascending (oldest message first), or -1 for descending (newest
            message first).
        :param limit: (Default None) The maximum number of messages
            to list. The results may include fewer messages than the
            requested `limit` if not enough are available. If limit is
            not specified

        :returns: Generator yielding up to `limit` messages.
        """

        if sort not in (1, -1):
            raise ValueError(u'sort must be either 1 (ascending) '
                             u'or -1 (descending)')

        now = timeutils.utcnow_ts()

        query = {
            # Messages must belong to this queue and project.
            PROJ_QUEUE: utils.scope_queue_name(queue_name, project),

            # NOTE(kgriffs): Messages must be finalized (i.e., must not
            # be part of an unfinalized transaction).
            #
            # See also the note wrt 'tx' within the definition
            # of ACTIVE_INDEX_FIELDS.
            'tx': None,
        }

        if not echo:
            query['u'] = {'$ne': client_uuid}

        if marker is not None:
            query['k'] = {'$gt': marker}

        collection = self._collection(queue_name, project)

        if not include_claimed:
            # Only include messages that are not part of
            # any claim, or are part of an expired claim.
            query['c.e'] = {'$lte': now}

        if not include_delayed:
            # NOTE(cdyangzhenyu): Only include messages that are not
            # part of any delay, or are part of an expired delay. if
            # the message has no attribute 'd', it will also be obtained.
            # This is for compatibility with old data.
            query['$or'] = [{'d': {'$lte': now}},
                            {'d': {'$exists': False}}]

        # Construct the request
        cursor = collection.find(query,
                                 projection=projection,
                                 sort=[('k', sort)])

        if limit is not None:
            cursor.limit(limit)

        # NOTE(flaper87): Suggest the index to use for this query to
        # ensure the most performant one is chosen.
        return cursor.hint(ACTIVE_INDEX_FIELDS)

    def create(self, queue, metadata, project=None,
               limit=storage.DEFAULT_MESSAGES_PER_CLAIM):
        """Creates a claim.

        This implementation was done in a best-effort fashion.
        In order to create a claim we need to get a list
        of messages that can be claimed. Once we have that
        list we execute a query filtering by the ids returned
        by the previous query.

        Since there's a lot of space for race conditions here,
        we'll check if the number of updated records is equal to
        the max number of messages to claim. If the number of updated
        messages is lower than limit we'll try to claim the remaining
        number of messages.

        This 2 queries are required because there's no way, as for the
        time being, to execute an update on a limited number of records.
        """
        msg_ctrl = self.driver.message_controller

        ttl = metadata['ttl']
        grace = metadata['grace']
        oid = objectid.ObjectId()

        now = timeutils.utcnow_ts()
        claim_expires = now + ttl
        claim_expires_dt = datetime.datetime.utcfromtimestamp(claim_expires)

        message_ttl = ttl + grace
        message_expiration = datetime.datetime.utcfromtimestamp(
            claim_expires + grace)

        meta = {
            'id': oid,
            't': ttl,
            'e': claim_expires,
        }

        # Get a list of active, not claimed nor expired
        # messages that could be claimed.
        msgs = msg_ctrl._active(queue, projection={'_id': 1}, project=project,
                                limit=limit)

        messages = iter([])
        ids = [msg['_id'] for msg in msgs]

        if len(ids) == 0:
            return None, messages

        now = timeutils.utcnow_ts()

        # NOTE(kgriffs): Set the claim field for
        # the active message batch, while also
        # filtering out any messages that happened
        # to get claimed just now by one or more
        # parallel requests.
        #
        # Filtering by just 'c.e' works because
        # new messages have that field initialized
        # to the current time when the message is
        # posted. There is no need to check whether
        # 'c' exists or 'c.id' is None.
        collection = msg_ctrl._collection(queue, project)
        updated = collection.update({'_id': {'$in': ids},
                                     'c.e': {'$lte': now}},
                                    {'$set': {'c': meta}},
                                    upsert=False,
                                    multi=True)['n']

        # NOTE(flaper87): Dirty hack!
        # This sets the expiration time to
        # `expires` on messages that would
        # expire before claim.
        new_values = {'e': message_expiration, 't': message_ttl}
        collection.update({'p_q': utils.scope_queue_name(queue, project),
                           'e': {'$lt': claim_expires_dt},
                           'c.id': oid},
                          {'$set': new_values},
                          upsert=False, multi=True)

        if updated != 0:
            # NOTE(kgriffs): This extra step is necessary because
            # in between having gotten a list of active messages
            # and updating them, some of them may have been
            # claimed by a parallel request. Therefore, we need
            # to find out which messages were actually tagged
            # with the claim ID successfully.
            claim, messages = self.get(queue, oid, project=project)

        return str(oid), messages