def on_put(self, req, resp, project_id, queue_name, subscription_id):
        if req.content_length:
            document = wsgi_utils.deserialize(req.stream, req.content_length)
        else:
            document = {}

        try:
            self._validate.subscription_confirming(document)
            confirm = document.get('confirmed', None)
            self._subscription_controller.confirm(queue_name, subscription_id,
                                                  project=project_id,
                                                  confirm=confirm)
            resp.status = falcon.HTTP_204
            resp.location = req.path
        except storage_errors.SubscriptionDoesNotExist as ex:
            LOG.debug(ex)
            raise wsgi_errors.HTTPNotFound(six.text_type(ex))
        except validation.ValidationFailed as ex:
            LOG.debug(ex)
            raise wsgi_errors.HTTPBadRequestAPI(six.text_type(ex))
        except Exception as ex:
            LOG.exception(ex)
            description = (_(u'Subscription %(subscription_id)s could not be'
                             ' confirmed.') %
                           dict(subscription_id=subscription_id))
            raise falcon.HTTPBadRequest(_('Unable to confirm subscription'),
                                        description)

def require_client_id(validate, req, resp, params):
    """Makes sure the header `Client-ID` is present in the request

    Use as a before hook.
    :param validate: A validator function that will
        be used to check the format of client id against configured
        limits.
    :param req: request sent
    :type req: falcon.request.Request
    :param resp: response object to return
    :type resp: falcon.response.Response
    :param params: additional parameters passed to responders
    :type params: dict
    :rtype: None
    """

    if req.path.startswith('/v1.1/') or req.path.startswith('/v2/'):
        try:
            validate(req.get_header('Client-ID', required=True))
        except ValueError:
            description = _(u'Malformed hexadecimal UUID.')
            raise falcon.HTTPBadRequest('Wrong UUID value', description)
        except validation.ValidationFailed as ex:
            raise falcon.HTTPBadRequest(six.text_type(ex))
    else:
        # NOTE(wanghao): Since we changed the get_client_uuid to support
        # other format of client id, so need to check the uuid here for
        # v1 API.
        try:
            client_id = req.get_header('Client-ID')
            if client_id or client_id == '':
                uuid.UUID(client_id)
        except ValueError:
            description = _(u'Malformed hexadecimal UUID.')
            raise falcon.HTTPBadRequest('Wrong UUID value', description)

    def queue_create(self, req):
        """Creates a queue

        :param req: Request instance ready to be sent.
        :type req: `api.common.Request`
        :return: resp: Response instance
        :type: resp: `api.common.Response`
        """
        project_id = req._headers.get('X-Project-ID')
        queue_name = req._body.get('queue_name')
        metadata = req._body.get('metadata', {})

        LOG.debug(u'Queue create - queue: %(queue)s, project: %(project)s',
                  {'queue': queue_name,
                   'project': project_id})

        try:
            self._validate.queue_identification(queue_name, project_id)
            self._validate.queue_metadata_length(len(str(metadata)))
            created = self._queue_controller.create(queue_name,
                                                    metadata=metadata,
                                                    project=project_id)
        except validation.ValidationFailed as ex:
            LOG.debug(ex)
            headers = {'status': 400}
            return api_utils.error_response(req, ex, headers)
        except storage_errors.ExceptionBase as ex:
            LOG.exception(ex)
            error = _('Queue %s could not be created.') % queue_name
            headers = {'status': 503}
            return api_utils.error_response(req, ex, headers, error)
        else:
            body = _('Queue %s created.') % queue_name
            headers = {'status': 201} if created else {'status': 204}
            return response.Response(req, body, headers)

    def on_put(self, request, response, project_id, flavor):
        """Registers a new flavor. Expects the following input:

        ::

            {"pool": "my-pool", "capabilities": {}}

        A capabilities object may also be provided.

        :returns: HTTP | [201, 400]
        """

        LOG.debug(u"PUT flavor - name: %s", flavor)

        data = wsgi_utils.load(request)
        wsgi_utils.validate(self._validators["create"], data)

        try:
            self._ctrl.create(flavor, pool=data["pool"], project=project_id)
            response.status = falcon.HTTP_201
            response.location = request.path
        except errors.PoolDoesNotExist as ex:
            LOG.exception(ex)
            description = _(u"Flavor %(flavor)s could not be created. " u"Pool %(pool)s does not exist") % dict(
                flavor=flavor, pool=data["pool"]
            )
            raise falcon.HTTPBadRequest(_("Unable to create"), description)

    def __init__(self, conf, cache):
        super(DataDriver, self).__init__(conf, cache)

        self.mongodb_conf = self.conf[options.MONGODB_GROUP]

        conn = self.connection
        server_version = conn.server_info()['version']

        if tuple(map(int, server_version.split('.'))) < (2, 2):
            raise RuntimeError(_('The mongodb driver requires mongodb>=2.2,  '
                                 '%s found') % server_version)

        if not len(conn.nodes) > 1 and not conn.is_mongos:
            if not self.conf.unreliable:
                raise RuntimeError(_('Either a replica set or a mongos is '
                                     'required to guarantee message delivery'))
        else:
            wc = conn.write_concern.get('w')
            majority = (wc == 'majority' or
                        wc >= 2)

            if not wc:
                # NOTE(flaper87): No write concern specified, use majority
                # and don't count journal as a replica. Use `update` to avoid
                # overwriting `wtimeout`
                conn.write_concern.update({'w': 'majority'})
            elif not self.conf.unreliable and not majority:
                raise RuntimeError(_('Using a write concern other than '
                                     '`majority` or > 2 makes the service '
                                     'unreliable. Please use a different '
                                     'write concern or set `unreliable` '
                                     'to True in the config file.'))

            conn.write_concern['j'] = False

def get_checked_field(document, name, value_type, default_value):
    """Validates and retrieves a typed field from a document.

    This function attempts to look up doc[name], and raises
    appropriate HTTP errors if the field is missing or not an
    instance of the given type.

    :param document: dict-like object
    :param name: field name
    :param value_type: expected value type, or '*' to accept any type
    :param default_value: Default value to use if the value is missing,
        or None to make the value required.
    :raises HTTPBadRequest: if the field is missing or not an
        instance of value_type
    :returns: value obtained from doc[name]
    """

    try:
        value = document[name]
    except KeyError:
        if default_value is not None:
            value = default_value
        else:
            description = _(u'Missing "{name}" field.').format(name=name)
            raise errors.HTTPBadRequestBody(description)

    # PERF(kgriffs): We do our own little spec thing because it is way
    # faster than jsonschema.
    if value_type == '*' or isinstance(value, value_type):
        return value

    description = _(u'The value of the "{name}" field must be a {vtype}.')
    description = description.format(name=name, vtype=value_type.__name__)
    raise errors.HTTPBadRequestBody(description)

    def on_delete(self, req, resp, project_id, queue_name, message_id):
        LOG.debug(u'Messages item DELETE - message: %(message)s, '
                  u'queue: %(queue)s, project: %(project)s',
                  {'message': message_id,
                   'queue': queue_name,
                   'project': project_id})
        try:
            self.message_controller.delete(
                queue_name,
                message_id=message_id,
                project=project_id,
                claim=req.get_param('claim_id'))

        except storage_errors.NotPermitted as ex:
            LOG.exception(ex)
            title = _(u'Unable to delete')
            description = _(u'This message is claimed; it cannot be '
                            u'deleted without a valid claim_id.')
            raise falcon.HTTPForbidden(title, description)

        except Exception as ex:
            LOG.exception(ex)
            description = _(u'Message could not be deleted.')
            raise wsgi_errors.HTTPServiceUnavailable(description)

        # Alles guete
        resp.status = falcon.HTTP_204

    def queue_delete(self, req):
        """Deletes a queue

        :param req: Request instance ready to be sent.
        :type req: `api.common.Request`
        :return: resp: Response instance
        :type: resp: `api.common.Response`
        """
        project_id = req._headers.get('X-Project-ID')
        queue_name = req._body.get('queue_name')

        LOG.debug(u'Queue delete - queue: %(queue)s, project: %(project)s',
                  {'queue': queue_name, 'project': project_id})
        try:
            self._queue_controller.delete(queue_name, project=project_id)
        except storage_errors.ExceptionBase as ex:
            LOG.exception(ex)
            error = _('Queue %s could not be deleted.') % queue_name
            headers = {'status': 503}
            return api_utils.error_response(req, ex, headers, error)
        else:
            body = _('Queue %s removed.') % queue_name
            headers = {'status': 204}
            resp = response.Response(req, body, headers)
            return resp

    def delete(self, project, queue):
        # (gengchc): Check if the queue already exists.
        if not self._exists(project, queue):
            return True

        queue_key = utils.scope_queue_name(queue, project)
        catalogue_project_key = utils.scope_pool_catalogue(project,
                                                           CATALOGUE_SUFFIX)
        catalogue_queue_key = utils.scope_pool_catalogue(queue_key,
                                                         CATALOGUE_SUFFIX)
        # (gengchc) Pipeline ensures atomic inserts.
        with self._client.pipeline() as pipe:
            pipe.zrem(catalogue_project_key, queue_key)
            pipe.delete(catalogue_queue_key)
            try:
                pipe.execute()
            except redis.exceptions.ResponseError:
                msgtmpl = _(u'CatalogueController:delete %(prj)s'
                            ':%(queue)s failed')
                LOG.info(msgtmpl,
                         {'prj': project, 'queue': queue})
                return False
        msgtmpl = _(u'CatalogueController:delete %(prj)s:%(queue)s success')
        LOG.info(msgtmpl,
                 {'prj': project, 'queue': queue})

    def queue_get(self, req):
        """Gets a queue

        :param req: Request instance ready to be sent.
        :type req: `api.common.Request`
        :return: resp: Response instance
        :type: resp: `api.common.Response`
        """
        project_id = req._headers.get('X-Project-ID')
        queue_name = req._body.get('queue_name')

        LOG.debug(u'Queue get - queue: %(queue)s, '
                  u'project: %(project)s',
                  {'queue': queue_name, 'project': project_id})

        try:
            resp_dict = self._queue_controller.get(queue_name,
                                                   project=project_id)
        except storage_errors.DoesNotExist as ex:
            LOG.debug(ex)
            error = _('Queue %s does not exist.') % queue_name
            headers = {'status': 404}
            return api_utils.error_response(req, ex, headers, error)
        except storage_errors.ExceptionBase as ex:
            LOG.exception(ex)
            headers = {'status': 503}
            error = _('Cannot retrieve queue %s.') % queue_name
            return api_utils.error_response(req, ex, headers, error)
        else:
            body = resp_dict
            headers = {'status': 200}
            return response.Response(req, body, headers)

    def __init__(self, conf, cache, control_driver):
        super(DataDriver, self).__init__(conf, cache, control_driver)

        self.mongodb_conf = self.conf[options.MESSAGE_MONGODB_GROUP]

        conn = self.connection
        server_info = conn.server_info()['version']
        self.server_version = tuple(map(int, server_info.split('.')))

        if self.server_version < (2, 2):
            raise RuntimeError(_('The mongodb driver requires mongodb>=2.2, '
                                 '%s found') % server_info)

        if not len(conn.nodes) > 1 and not conn.is_mongos:
            if not self.conf.unreliable:
                raise RuntimeError(_('Either a replica set or a mongos is '
                                     'required to guarantee message delivery'))
        else:

            _mongo_wc = conn.write_concern.document.get('w')
            durable = (_mongo_wc == 'majority' or
                       _mongo_wc >= 2)

            if not self.conf.unreliable and not durable:
                raise RuntimeError(_('Using a write concern other than '
                                     '`majority` or > 2 makes the service '
                                     'unreliable. Please use a different '
                                     'write concern or set `unreliable` '
                                     'to True in the config file.'))

        # FIXME(flaper87): Make this dynamic
        self._capabilities = self.BASE_CAPABILITIES

def require_content_type_be_non_urlencoded(req, resp, params):
    """Raises an exception on "x-www-form-urlencoded" content type of request.

    If request has body and "Content-Type" header has
    "application/x-www-form-urlencoded" value (case-insensitive), this function
    raises falcon.HTTPBadRequest exception.

    This strange function exists only to prevent bug/1547100 in a backward
    compatible way.

    Meant to be used as a `before` hook.

    :param req: request sent
    :type req: falcon.request.Request
    :param resp: response object to return
    :type resp: falcon.response.Response
    :param params: additional parameters passed to responders
    :type params: dict
    :rtype: None
    :raises: falcon.HTTPBadRequest

    """
    if req.content_length is None:
        return
    if req.content_type and (req.content_type.lower() ==
                             'application/x-www-form-urlencoded'):
        title = _(u'Invalid Content-Type')
        description = _(u'Endpoint does not accept '
                        u'`application/x-www-form-urlencoded` content; '
                        u'currently supported media type is '
                        u'`application/json`; specify proper client-side '
                        u'media type with the "Content-Type" header.')
        raise falcon.HTTPBadRequest(title, description)

    def subscription_patching(self, subscription):
        """Restrictions on an update of subscription.

        :param subscription: dict of subscription
        :raises: ValidationFailed if the subscription is invalid.
        """

        if not subscription:
            raise ValidationFailed(_(u'No subscription to create.'))

        subscriber = subscription.get('subscriber', None)
        subscriber_type = None

        if subscriber:
            parsed_uri = six.moves.urllib_parse.urlparse(subscriber)
            subscriber_type = parsed_uri.scheme

            if subscriber_type not in self._limits_conf.subscriber_types:
                msg = _(u'The subscriber type of subscription must be '
                        u'supported in the list {0}.')
                raise ValidationFailed(msg, self._limits_conf.subscriber_types)

        options = subscription.get('options', None)
        if options and not isinstance(options, dict):
            msg = _(u'Options must be a dict.')
            raise ValidationFailed(msg)

        ttl = subscription.get('ttl', None)
        if ttl and not isinstance(ttl, int):
            msg = _(u'TTL must be an integer.')
            raise ValidationFailed(msg)

    def _insert(self, project, queue, pool):
        queue_key = utils.scope_queue_name(queue, project)
        catalogue_project_key = utils.scope_pool_catalogue(project,
                                                           CATALOGUE_SUFFIX)
        catalogue_queue_key = utils.scope_pool_catalogue(queue_key,
                                                         CATALOGUE_SUFFIX)
        # Check if the queue already exists.
        if self._exists(queue, project):
            return False

        catalogue = {
            'p': project,
            'p_q': queue,
            'p_p': pool
        }
        # Pipeline ensures atomic inserts.
        with self._client.pipeline() as pipe:
            pipe.zadd(catalogue_project_key, {queue_key: 1})
            pipe.hmset(catalogue_queue_key, catalogue)

            try:
                pipe.execute()
            except redis.exceptions.ResponseError:
                msgtmpl = _(u'CatalogueController:insert %(prj)s:'
                            '%(queue)s  %(pool)s failed')
                LOG.exception(msgtmpl,
                              {'prj': project, 'queue': queue, 'pool': pool})
                return False
        msgtmpl = _(u'CatalogueController:insert %(prj)s:%(queue)s'
                    ':%(pool)s, success')
        LOG.info(msgtmpl,
                 {'prj': project, 'queue': queue, 'pool': pool})
        return True

    def claim_creation(self, metadata, limit=None):
        """Restrictions on the claim parameters upon creation.

        :param metadata: The claim metadata
        :param limit: The number of messages to claim
        :raises: ValidationFailed if either TTL or grace is out of range,
            or the expected number of messages exceed the limit.
        """

        self.claim_updating(metadata)

        uplimit = self._limits_conf.max_messages_per_claim_or_pop
        if limit is not None and not (0 < limit <= uplimit):
            msg = _(u'Limit must be at least 1 and may not '
                    'be greater than {0}.')

            raise ValidationFailed(
                msg, self._limits_conf.max_messages_per_claim_or_pop)

        grace = metadata['grace']

        if not (MIN_CLAIM_GRACE <= grace <= self._limits_conf.max_claim_grace):
            msg = _(u'The grace for a claim may not exceed {0} seconds, and '
                    'must be at least {1} seconds long.')

            raise ValidationFailed(
                msg, self._limits_conf.max_claim_grace, MIN_CLAIM_GRACE)

    def message_length(self, content_length, max_msg_post_size=None):
        """Restrictions on message post length.

        :param content_length: Queue request's length.
        :raises: ValidationFailed if the metadata is oversize.
        """
        if content_length is None:
            return

        if max_msg_post_size:
            try:
                min_max_size = min(max_msg_post_size,
                                   self._limits_conf.max_messages_post_size)
                if content_length > min_max_size:
                    raise ValidationFailed(
                        _(u'Message collection size is too large. The max '
                          'size for current queue is {0}. It is calculated '
                          'by max size = min(max_messages_post_size_config: '
                          '{1}, max_messages_post_size_queue: {2}).'),
                        min_max_size,
                        self._limits_conf.max_messages_post_size,
                        max_msg_post_size)
            except TypeError:
                # NOTE(flwang): If there is a type error when using min(),
                # it only happens in py3.x, it will be skipped and compare
                # the message length with the size defined in config file.
                pass

        if content_length > self._limits_conf.max_messages_post_size:
            raise ValidationFailed(
                _(u'Message collection size is too large. Max size {0}'),
                self._limits_conf.max_messages_post_size)

    def queue_metadata_putting(self, queue_metadata):
        """Checking if the reserved attributes of the queue are valid.

        :param queue_metadata: Queue's metadata.
        :raises: ValidationFailed if any reserved attribute is invalid.
        """
        if not queue_metadata:
            return

        queue_default_ttl = queue_metadata.get('_default_message_ttl', None)
        if queue_default_ttl and not isinstance(queue_default_ttl, int):
            msg = _(u'_default_message_ttl must be integer.')
            raise ValidationFailed(msg)

        if queue_default_ttl:
            if not (MIN_MESSAGE_TTL <= queue_default_ttl <=
                    self._limits_conf.max_message_ttl):
                msg = _(u'_default_message_ttl can not exceed {0} '
                        'seconds, and must be at least {1} seconds long.')
                raise ValidationFailed(
                    msg, self._limits_conf.max_message_ttl, MIN_MESSAGE_TTL)

        queue_max_msg_size = queue_metadata.get('_max_messages_post_size',
                                                None)
        if queue_max_msg_size and not isinstance(queue_max_msg_size, int):
            msg = _(u'_max_messages_post_size must be integer.')
            raise ValidationFailed(msg)

        if queue_max_msg_size:
            if not (0 < queue_max_msg_size <=
                    self._limits_conf.max_messages_post_size):
                raise ValidationFailed(
                    _(u'_max_messages_post_size can not exceed {0}, '
                      ' and must be at least greater than 0.'),
                    self._limits_conf.max_messages_post_size)

    def queue_patching(self, request, changes):
        washed_changes = []
        content_types = {
            'application/openstack-messaging-v2.0-json-patch': 10,
        }

        json_schema_version = content_types[request.content_type]

        if not isinstance(changes, list):
            msg = _('Request body must be a JSON array of operation objects.')
            raise ValidationFailed(msg)

        for raw_change in changes:
            if not isinstance(raw_change, dict):
                msg = _('Operations must be JSON objects.')
                raise ValidationFailed(msg)

            (op, path) = self._parse_json_schema_change(raw_change,
                                                        json_schema_version)

            # NOTE(flwang): Now the 'path' is a list.
            self._validate_path(op, path)
            change = {'op': op, 'path': path,
                      'json_schema_version': json_schema_version}

            if not op == 'remove':
                change['value'] = self._get_change_value(raw_change, op)

            self._validate_change(change)

            washed_changes.append(change)

        return washed_changes

def extract_project_id(req, resp, params):
    """Adds `project_id` to the list of params for all responders

    Meant to be used as a `before` hook.

    :param req: request sent
    :type req: falcon.request.Request
    :param resp: response object to return
    :type resp: falcon.response.Response
    :param params: additional parameters passed to responders
    :type params: dict
    :rtype: None
    """
    api_version_string = req.path.split('/')[1]
    params['project_id'] = req.get_header('X-PROJECT-ID')
    if not api_version_string:
        # NOTE(jaosorior): The versions resource is public and shouldn't need
        # a check for the project-id.
        return
    if params['project_id'] == "":
        raise falcon.HTTPBadRequest('Empty project header not allowed',
                                    _(u'X-PROJECT-ID cannot be an empty '
                                      u'string. Specify the right header '
                                      u'X-PROJECT-ID and retry.'))

    api_version = version.LooseVersion(api_version_string)
    if (not params['project_id'] and api_version >=
            version.LooseVersion('v1.1')):
        raise falcon.HTTPBadRequest('Project-Id Missing',
                                    _(u'The header X-PROJECT-ID was missing'))

    def on_put(self, request, response, project_id, flavor):
        """Registers a new flavor. Expects the following input:

        ::

            {"pool": "my-pool", "capabilities": {}}

        A capabilities object may also be provided.

        :returns: HTTP | [201, 400]
        """

        LOG.debug(u'PUT flavor - name: %s', flavor)

        data = wsgi_utils.load(request)
        wsgi_utils.validate(self._validators['create'], data)

        try:
            self._ctrl.create(flavor,
                              pool=data['pool'],
                              project=project_id,
                              capabilities=data['capabilities'])
            response.status = falcon.HTTP_201
            response.location = request.path
        except errors.PoolDoesNotExist as ex:
            LOG.exception(ex)
            description = (_(u'Flavor {flavor} could not be created. '
                             u'Pool {pool} does not exist') %
                           dict(flavor=flavor, pool=data['pool']))
            raise falcon.HTTPBadRequest(_('Unable to create'), description)