def claimed(self, queue_name, claim_id=None,
                expires=None, limit=None, project=None):
        query = {
            'c.id': claim_id,
            'c.e': {'$gt': expires or timeutils.utcnow()},
            'q': queue_name,
            'p': project,
        }

        if not claim_id:
            # lookup over c.id to use the index
            query['c.id'] = {'$ne': None}

        msgs = self._col.find(query, sort=[('k', 1)])

        if limit:
            msgs = msgs.limit(limit)

        now = timeutils.utcnow()

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

            return doc

        return utils.HookedCursor(msgs, denormalizer)

    def claimed(self, queue_id, claim_id=None, expires=None, limit=None):
        query = {
            'c.id': claim_id,
            'c.e': {'$gt': expires or timeutils.utcnow()},
            'q': utils.to_oid(queue_id),
        }

        if not claim_id:
            # lookup over c.id to use the index
            query['c.id'] = {'$ne': None}

        msgs = self._col.find(query, sort=[('_id', 1)])

        if limit:
            msgs = msgs.limit(limit)

        now = timeutils.utcnow()

        def denormalizer(msg):
            oid = msg['_id']
            age = now - utils.oid_utc(oid)

            return {
                'id': str(oid),
                'age': age.seconds,
                'ttl': msg['t'],
                'body': msg['b'],
                'claim': msg['c']
            }

        return utils.HookedCursor(msgs, denormalizer)

    def claimed(self, queue_name, claim_id,
                expires=None, limit=None, project=None):
        query = {
            'c.id': claim_id,
            'c.e': {'$gt': expires or timeutils.utcnow()},
            'q': queue_name,
            'p': project,
        }

        preference = pymongo.read_preferences.ReadPreference.PRIMARY
        msgs = self._col.find(query, sort=[('k', 1)],
                              read_preference=preference)

        if limit:
            msgs = msgs.limit(limit)

        now = timeutils.utcnow()

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

            return doc

        return utils.HookedCursor(msgs, denormalizer)

    def get(self, queue, claim_id, project=None):
        if project is None:
            project = ''

        cid = utils.cid_decode(claim_id)
        if cid is None:
            raise errors.ClaimDoesNotExist(claim_id, queue, project)

        with self.driver.trans() as trans:
            sel = sa.sql.select([tables.Claims.c.id,
                                 tables.Claims.c.ttl,
                                 tables.Claims.c.created],
                                sa.and_(tables.Claims.c.ttl >
                                        utils.get_age(tables.Claims.c.created),
                                        tables.Claims.c.id == cid,
                                        tables.Queues.c.project == project,
                                        tables.Queues.c.name == queue),
                                from_obj=[tables.Queues.join(tables.Claims)])

            res = trans.execute(sel).fetchone()
            if res is None:
                raise errors.ClaimDoesNotExist(claim_id, queue, project)

            cid, ttl, created = res
            return (
                {'id': claim_id,
                 'ttl': ttl,
                 'age': (timeutils.utcnow() - created).seconds},
                self.__get(cid)
            )

    def get(self, queue_name, message_id, project=None):
        """Gets a single message by ID.

        :raises: exceptions.MessageDoesNotExist
        """
        mid = utils.to_oid(message_id)
        if mid is None:
            raise exceptions.MessageDoesNotExist(message_id, queue_name,
                                                 project)

        now = timeutils.utcnow()

        query = {
            '_id': mid,
            'q': queue_name,
            'p': project,
            'e': {'$gt': now}
        }

        message = list(self._col.find(query).limit(1).hint([('_id', 1)]))

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

        return _basic_message(message[0], now)

    def list(self, queue, project=None, marker=None,
             limit=10, echo=False, client_uuid=None):

        if marker is not None:
            try:
                marker = int(marker)
            except ValueError:
                raise exceptions.MalformedMarker()

        qid = self._get_queue_id(queue, project)
        messages = self.active(qid, marker, echo, client_uuid)

        messages = messages.limit(limit).sort('_id')
        marker_id = {}

        now = timeutils.utcnow()

        def denormalizer(msg):
            oid = msg['_id']
            age = now - utils.oid_utc(oid)
            marker_id['next'] = msg['k']

            return {
                'id': str(oid),
                'age': age.seconds,
                'ttl': msg['t'],
                'body': msg['b'],
            }

        yield utils.HookedCursor(messages, denormalizer)
        yield str(marker_id['next'])

    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

        query = {"q": queue_name, "p": project, "_id": mid}

        # NOTE(cpp-cabrera): return early - the user gaves us an
        # invalid claim id and that renders the rest of this
        # request moot
        cid = utils.to_oid(claim)
        if cid is None:
            return

        now = timeutils.utcnow()
        query["e"] = {"$gt": now}
        message = self._col.find_one(query)

        if message is None:
            return

        is_claimed = message["c"]["id"] is not None and message["c"]["e"] > now

        if claim is None:
            if is_claimed:
                raise exceptions.MessageIsClaimed(message_id)

        else:
            if message["c"]["id"] != cid:
                raise exceptions.MessageIsClaimedBy(message_id, claim)

        self._col.remove(query["_id"], w=0)

    def list(
        self, queue_name, project=None, marker=None, limit=None, echo=False, client_uuid=None, include_claimed=False
    ):

        if limit is None:
            limit = CFG.default_message_paging

        if marker is not None:
            try:
                marker = int(marker)
            except ValueError:
                yield iter([])

        messages = self._list(queue_name, marker, echo, client_uuid, include_claimed=include_claimed, project=project)

        messages = messages.limit(limit)
        marker_id = {}

        now = timeutils.utcnow()

        def denormalizer(msg):
            marker_id["next"] = msg["k"]

            return _basic_message(msg, now)

        yield utils.HookedCursor(messages, denormalizer)
        yield str(marker_id["next"])

    def get(self, queue, message_ids, project=None):
        if not isinstance(message_ids, list):
            message_ids = [message_ids]

        message_ids = [utils.to_oid(id) for id in message_ids]
        now = timeutils.utcnow()

        # Base query, always check expire time
        query = {
            'q': self._get_queue_id(queue, project),
            'e': {'$gt': now},
            '_id': {'$in': message_ids},
        }

        messages = self._col.find(query)

        def denormalizer(msg):
            oid = msg['_id']
            age = now - utils.oid_utc(oid)

            return {
                'id': str(oid),
                'age': age.seconds,
                'ttl': msg['t'],
                'body': msg['b'],
            }

        return utils.HookedCursor(messages, denormalizer)

    def _test_post(self, sample_messages):
        sample_doc = jsonutils.dumps(sample_messages)

        result = self.simulate_post(self.messages_path, self.project_id,
                                    body=sample_doc, headers=self.headers)
        self.assertEqual(self.srmock.status, falcon.HTTP_201)

        result_doc = jsonutils.loads(result[0])

        msg_ids = self._get_msg_ids(self.srmock.headers_dict)
        self.assertEqual(len(msg_ids), len(sample_messages))

        expected_resources = [six.text_type(self.messages_path + '/' + id)
                              for id in msg_ids]
        self.assertEqual(expected_resources, result_doc['resources'])
        self.assertFalse(result_doc['partial'])

        self.assertEqual(len(msg_ids), len(sample_messages))

        lookup = dict([(m['ttl'], m['body']) for m in sample_messages])

        # Test GET on the message resource directly
        # NOTE(cpp-cabrera): force the passing of time to age a message
        timeutils_utcnow = 'marconi.openstack.common.timeutils.utcnow'
        now = timeutils.utcnow() + datetime.timedelta(seconds=10)
        with mock.patch(timeutils_utcnow) as mock_utcnow:
            mock_utcnow.return_value = now
            for msg_id in msg_ids:
                message_uri = self.messages_path + '/' + msg_id

                # Wrong project ID
                self.simulate_get(message_uri, '777777')
                self.assertEqual(self.srmock.status, falcon.HTTP_404)

                # Correct project ID
                result = self.simulate_get(message_uri, self.project_id)
                self.assertEqual(self.srmock.status, falcon.HTTP_200)
                self.assertEqual(self.srmock.headers_dict['Content-Location'],
                                 message_uri)

                # Check message properties
                message = jsonutils.loads(result[0])
                self.assertEqual(message['href'], message_uri)
                self.assertEqual(message['body'], lookup[message['ttl']])

                # no negative age
                # NOTE(cpp-cabrera): testtools lacks GreaterThanEqual on py26
                self.assertThat(message['age'],
                                matchers.GreaterThan(-1))

        # Test bulk GET
        query_string = 'ids=' + ','.join(msg_ids)
        result = self.simulate_get(self.messages_path, self.project_id,
                                   query_string=query_string)

        self.assertEqual(self.srmock.status, falcon.HTTP_200)
        result_doc = jsonutils.loads(result[0])
        expected_ttls = set(m['ttl'] for m in sample_messages)
        actual_ttls = set(m['ttl'] for m in result_doc)
        self.assertFalse(expected_ttls - actual_ttls)

    def delete(self, queue, message_id, project=None, claim=None):
        try:
            mid = utils.to_oid(message_id)

            query = {
                'q': self._get_queue_id(queue, project),
                '_id': mid
            }

            if claim:
                now = timeutils.utcnow()
                query['e'] = {'$gt': now}
                message = self._col.find_one(query)

                if message is None:
                    return

                cid = utils.to_oid(claim)

                if not ('c' in message and
                        message['c']['id'] == cid and
                        message['c']['e'] > now):
                    raise exceptions.ClaimNotPermitted(message_id, claim)

                self._col.remove(query['_id'], w=0)
            else:
                self._col.remove(query, w=0)
        except exceptions.QueueDoesNotExist:
            pass

    def stats(self, name, project=None):
        if not self.exists(name, project=project):
            raise exceptions.QueueDoesNotExist(name, project)

        controller = self.driver.message_controller

        active = controller.active(name, project=project).count()
        total = controller.count(name, project=project)

        message_stats = {
            'claimed': total - active,
            'free': active,
            'total': total,
        }

        try:
            oldest = controller.first(name, project=project, sort=1)
            newest = controller.first(name, project=project, sort=-1)
        except exceptions.QueueIsEmpty:
            pass
        else:
            now = timeutils.utcnow()
            message_stats['oldest'] = utils.stat_message(oldest, now)
            message_stats['newest'] = utils.stat_message(newest, now)

        return {'messages': message_stats}

    def _remove_expired(self, queue_id):
        """Removes all expired messages except for the most recent
        in each queue.

        This method is used in lieu of mongo's TTL index since we
        must always leave at least one message in the queue for
        calculating the next marker.

        Note that expired messages are only removed if their count
        exceeds options.CFG.gc_threshold.

        :param queue_id: id for the queue from which to remove
            expired messages
        """

        if options.CFG.gc_threshold <= self._count_expired(queue_id):
            # Get the message with the highest marker, and leave
            # it in the queue
            head = self._col.find_one({'q': queue_id},
                                      sort=[('k', -1)],
                                      fields={'_id': 1})

            if head is None:
                # Assume queue was just deleted via a parallel request
                LOG.warning(_('Queue %s is empty or missing.') % queue_id)
                return

            query = {
                'q': queue_id,
                'e': {'$lte': timeutils.utcnow()},
                '_id': {'$ne': head['_id']}
            }

            self._col.remove(query)

    def test_message_counter(self):
        queue_name = self.queue_name
        iterations = 10

        seed_marker1 = self.queue_controller._get_counter(queue_name,
                                                          self.project)
        self.assertEqual(seed_marker1, 1, 'First marker is 1')

        for i in range(iterations):
            self.controller.post(queue_name, [{'ttl': 60}],
                                 'uuid', project=self.project)

            marker1 = self.queue_controller._get_counter(queue_name,
                                                         self.project)
            marker2 = self.queue_controller._get_counter(queue_name,
                                                         self.project)
            marker3 = self.queue_controller._get_counter(queue_name,
                                                         self.project)

            self.assertEqual(marker1, marker2)
            self.assertEqual(marker2, marker3)
            self.assertEqual(marker1, i + 2)

        new_value = self.queue_controller._inc_counter(queue_name,
                                                       self.project)
        self.assertIsNotNone(new_value)

        value_before = self.queue_controller._get_counter(queue_name,
                                                          project=self.project)
        new_value = self.queue_controller._inc_counter(queue_name,
                                                       project=self.project)
        self.assertIsNotNone(new_value)
        value_after = self.queue_controller._get_counter(queue_name,
                                                         project=self.project)
        self.assertEqual(value_after, value_before + 1)

        value_before = value_after
        new_value = self.queue_controller._inc_counter(queue_name,
                                                       project=self.project,
                                                       amount=7)
        value_after = self.queue_controller._get_counter(queue_name,
                                                         project=self.project)
        self.assertEqual(value_after, value_before + 7)
        self.assertEqual(value_after, new_value)

        reference_value = value_after

        unchanged = self.queue_controller._inc_counter(queue_name,
                                                       project=self.project,
                                                       window=10)
        self.assertIsNone(unchanged)

        now = timeutils.utcnow() + datetime.timedelta(seconds=10)
        timeutils_utcnow = 'marconi.openstack.common.timeutils.utcnow'
        with mock.patch(timeutils_utcnow) as mock_utcnow:
            mock_utcnow.return_value = now
            changed = self.queue_controller._inc_counter(queue_name,
                                                         project=self.project,
                                                         window=5)
            self.assertEqual(changed, reference_value + 1)

    def _count_expired(self, queue_name, project=None):
        """Counts the number of expired messages in a queue.

        :param queue_name: Name of the queue to stat
        """

        query = {"p": project, "q": queue_name, "e": {"$lte": timeutils.utcnow()}}

        return self._col.find(query).count()

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

        :param queue_name: Name of the queue to list
        :param project: Project `queue_name` belongs to.
        :param marker: Message marker from which to start iterating
        :param echo: Whether to return messages that match client_uuid
        :param client_uuid: UUID for the client that originated this request
        :param fields: Fields to include in emmitted documents as a dict
        :param include_claimed: Whether to include claimed 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).

        :returns: MongoDB cursor
        """

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

        now = timeutils.utcnow()

        query = {
            # Messages must belong to this
            # queue and project
            "p": project,
            "q": queue_name,
            # The messages cannot be expired
            "e": {"$gt": now},
        }

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

        if marker:
            query["k"] = {"$gt": marker}

        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}

        # NOTE(flaper87): Suggest the index to use for this query
        return self._col.find(query, fields=fields, sort=[("k", sort)]).hint(self.active_fields)

    def _test_post(self, sample_messages):
        sample_doc = json.dumps(sample_messages)

        result = self.simulate_post(self.messages_path, self.project_id, body=sample_doc, headers=self.headers)
        self.assertEqual(self.srmock.status, falcon.HTTP_201)

        result_doc = json.loads(result[0])

        msg_ids = self._get_msg_ids(self.srmock.headers_dict)
        self.assertEqual(len(msg_ids), len(sample_messages))

        expected_resources = [unicode(self.messages_path + "/" + id) for id in msg_ids]
        self.assertEqual(expected_resources, result_doc["resources"])
        self.assertFalse(result_doc["partial"])

        self.assertEqual(len(msg_ids), len(sample_messages))

        lookup = dict([(m["ttl"], m["body"]) for m in sample_messages])

        # Test GET on the message resource directly
        # NOTE(cpp-cabrera): force the passing of time to age a message
        timeutils.set_time_override(timeutils.utcnow())
        timeutils.advance_time_seconds(10)
        for msg_id in msg_ids:
            message_uri = self.messages_path + "/" + msg_id

            # Wrong project ID
            self.simulate_get(message_uri, "777777")
            self.assertEqual(self.srmock.status, falcon.HTTP_404)

            # Correct project ID
            result = self.simulate_get(message_uri, self.project_id)
            self.assertEqual(self.srmock.status, falcon.HTTP_200)
            self.assertEqual(self.srmock.headers_dict["Content-Location"], message_uri)

            # Check message properties
            message = json.loads(result[0])
            self.assertEqual(message["href"], message_uri)
            self.assertEqual(message["body"], lookup[message["ttl"]])

            # no negative age
            # NOTE(cpp-cabrera): testtools lacks GreaterThanEqual on py26
            self.assertThat(message["age"], matchers.GreaterThan(-1))
        timeutils.clear_time_override()

        # Test bulk GET
        query_string = "ids=" + ",".join(msg_ids)
        result = self.simulate_get(self.messages_path, self.project_id, query_string=query_string)

        self.assertEqual(self.srmock.status, falcon.HTTP_200)
        result_doc = json.loads(result[0])
        expected_ttls = set(m["ttl"] for m in sample_messages)
        actual_ttls = set(m["ttl"] for m in result_doc)
        self.assertFalse(expected_ttls - actual_ttls)

    def claimed(self, queue_name, claim_id, expires=None, limit=None, project=None):
        query = {"c.id": claim_id, "c.e": {"$gt": expires or timeutils.utcnow()}, "q": queue_name, "p": project}

        # 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
        msgs = self._col.find(query, sort=[("k", 1)], read_preference=preference)

        if limit:
            msgs = msgs.limit(limit)

        now = timeutils.utcnow()

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

            return doc

        return utils.HookedCursor(msgs, denormalizer)

    def _count_expired(self, queue_id):
        """Counts the number of expired messages in a queue.

        :param queue_id: id for the queue to stat
        """

        query = {
            'q': queue_id,
            'e': {'$lte': timeutils.utcnow()},
        }

        return self._col.find(query).count()

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

        :param queue_name: Name of the queue to list
        :param project: Project `queue_name` belongs to.
        :param marker: Message marker from which to start iterating
        :param echo: Whether to return messages that match client_uuid
        :param client_uuid: UUID for the client that originated this request
        :param fields: Fields to include in emmitted documents
        :param include_claimed: Whether to include claimed 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).

        :returns: MongoDB cursor
        """

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

        now = timeutils.utcnow()

        query = {
            # Messages must belong to this
            # queue and project
            'p': project,
            'q': queue_name,

            # The messages cannot be expired
            'e': {'$gt': now},
        }

        if fields and not isinstance(fields, (dict, list)):
            raise TypeError('Fields must be an instance of list / dict')

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

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

        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}

        # NOTE(flaper87): Suggest the index to use for this query
        return self._col.find(query, fields=fields,
                              sort=[('k', sort)]).hint(self.active_fields)