<commit_before># Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import copy
import functools
import uuid

from oslo_serialization import jsonutils
from oslo_utils import timeutils
import swiftclient

from zaqar.common import decorators
from zaqar import storage
from zaqar.storage import errors
from zaqar.storage.swift import utils


class MessageController(storage.Message):
    """Implements message resource operations with swift backend

    Messages are scoped by project + queue.

    message -> Swift mapping:
       +--------------+-----------------------------------------+
       | Attribute    | Storage location                        |
       +--------------+-----------------------------------------+
       | Msg UUID     | Object name                             |
       +--------------+-----------------------------------------+
       | Queue Name   | Container name prefix                   |
       +--------------+-----------------------------------------+
       | Project name | Container name prefix                   |
       +--------------+-----------------------------------------+
       | Created time | Object Creation Time                    |
       +--------------+-----------------------------------------+
       | Msg Body     | Object content 'body'                   |
       +--------------+-----------------------------------------+
       | Client ID    | Object header 'ClientID'                |
       +--------------+-----------------------------------------+
       | Claim ID     | Object content 'claim_id'               |
       +--------------+-----------------------------------------+
       | Expires      | Object Delete-After header              |
       +--------------------------------------------------------+
    """

    def __init__(self, *args, **kwargs):
        super(MessageController, self).__init__(*args, **kwargs)
        self._client = self.driver.connection

    @decorators.lazy_property(write=False)
    def _queue_ctrl(self):
        return self.driver.queue_controller

    def _delete_queue_messages(self, queue, project, pipe):
        """Method to remove all the messages belonging to a queue.

        Will be referenced from the QueueController.
        The pipe to execute deletion will be passed from the QueueController
        executing the operation.
        """
        container = utils._message_container(queue, project)
        remaining = True
        key = ''
        while remaining:
            headers, objects = self._client.get_container(container,
                                                          limit=1000,
                                                          marker=key)
            if not objects:
                return
            remaining = len(objects) == 1000
            key = objects[-1]['name']
            for o in objects:
                try:
                    self._client.delete_object(container, o['name'])
                except swiftclient.ClientException as exc:
                    if exc.http_status == 404:
                        continue
                    raise

    def _list(self, queue, project=None, marker=None,
              limit=storage.DEFAULT_MESSAGES_PER_PAGE,
              echo=False, client_uuid=None,
              include_claimed=False, sort=1):
        """List messages in the queue, oldest first(ish)

        Time ordering and message inclusion in lists are soft, there is no
        global order and times are based on the UTC time of the zaqar-api
        server that the message was created from.

        Here be consistency dragons.
        """
        if not self._queue_ctrl.exists(queue, project):
            raise errors.QueueDoesNotExist(queue, project)

        client = self._client
        container = utils._message_container(queue, project)
        query_string = None
        if sort == -1:
            query_string = 'reverse=on'

        try:
            _, objects = client.get_container(
                container,
                marker=marker,
                # list 2x the objects because some listing items may have
                # expired
                limit=limit * 2,
                query_string=query_string)
        except swiftclient.ClientException as exc:
            if exc.http_status == 404:
                raise errors.QueueDoesNotExist(queue, project)
            raise

        def is_claimed(msg, headers):
            if include_claimed:
                return False
            return msg['claim_id'] is not None

        def is_echo(msg, headers):
            if echo:
                return False
            return headers['x-object-meta-clientid'] == str(client_uuid)

        filters = [
            is_echo,
            is_claimed,
        ]
        marker = {}
        get_object = functools.partial(client.get_object, container)
        list_objects = functools.partial(client.get_container, container,
                                         limit=limit * 2,
                                         query_string=query_string)
        yield utils._filter_messages(objects, filters, marker, get_object,
                                     list_objects, limit=limit)
        yield marker and marker['next']

    def list(self, queue, project=None, marker=None,<commit_msg>Fix message claim expires problem in swift storage<commit_after># Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import copy
import functools
import uuid

from oslo_serialization import jsonutils
from oslo_utils import timeutils
import swiftclient

from zaqar.common import decorators
from zaqar import storage
from zaqar.storage import errors
from zaqar.storage.swift import utils


class MessageController(storage.Message):
    """Implements message resource operations with swift backend

    Messages are scoped by project + queue.

    message -> Swift mapping:
       +--------------+-----------------------------------------+
       | Attribute    | Storage location                        |
       +--------------+-----------------------------------------+
       | Msg UUID     | Object name                             |
       +--------------+-----------------------------------------+
       | Queue Name   | Container name prefix                   |
       +--------------+-----------------------------------------+
       | Project name | Container name prefix                   |
       +--------------+-----------------------------------------+
       | Created time | Object Creation Time                    |
       +--------------+-----------------------------------------+
       | Msg Body     | Object content 'body'                   |
       +--------------+-----------------------------------------+
       | Client ID    | Object header 'ClientID'                |
       +--------------+-----------------------------------------+
       | Claim ID     | Object content 'claim_id'               |
       +--------------+-----------------------------------------+
       | Expires      | Object Delete-After header              |
       +--------------------------------------------------------+
    """

    def __init__(self, *args, **kwargs):
        super(MessageController, self).__init__(*args, **kwargs)
        self._client = self.driver.connection

    @decorators.lazy_property(write=False)
    def _queue_ctrl(self):
        return self.driver.queue_controller

    def _delete_queue_messages(self, queue, project, pipe):
        """Method to remove all the messages belonging to a queue.

        Will be referenced from the QueueController.
        The pipe to execute deletion will be passed from the QueueController
        executing the operation.
        """
        container = utils._message_container(queue, project)
        remaining = True
        key = ''
        while remaining:
            headers, objects = self._client.get_container(container,
                                                          limit=1000,
                                                          marker=key)
            if not objects:
                return
            remaining = len(objects) == 1000
            key = objects[-1]['name']
            for o in objects:
                try:
                    self._client.delete_object(container, o['name'])
                except swiftclient.ClientException as exc:
                    if exc.http_status == 404:
                        continue
                    raise

    def _list(self, queue, project=None, marker=None,
              limit=storage.DEFAULT_MESSAGES_PER_PAGE,
              echo=False, client_uuid=None,
              include_claimed=False, sort=1):
        """List messages in the queue, oldest first(ish)

        Time ordering and message inclusion in lists are soft, there is no
        global order and times are based on the UTC time of the zaqar-api
        server that the message was created from.

        Here be consistency dragons.
        """
        if not self._queue_ctrl.exists(queue, project):
            raise errors.QueueDoesNotExist(queue, project)

        client = self._client
        container = utils._message_container(queue, project)
        query_string = None
        if sort == -1:
            query_string = 'reverse=on'

        try:
            _, objects = client.get_container(
                container,
                marker=marker,
                # list 2x the objects because some listing items may have
                # expired
                limit=limit * 2,
                query_string=query_string)
        except swiftclient.ClientException as exc:
            if exc.http_status == 404:
                raise errors.QueueDoesNotExist(queue, project)
            raise

        def is_claimed(msg, headers):
            if include_claimed or msg['claim_id'] is None:
                return False
            claim_obj = self.driver.claim_controller._get(
                queue, msg['claim_id'], project)
            return claim_obj is not None and claim_obj['ttl'] > 0

        def is_echo(msg, headers):
            if echo:
                return False
            return headers['x-object-meta-clientid'] == str(client_uuid)

        filters = [
            is_echo,
            is_claimed,
        ]
        marker = {}
        get_object = functools.partial(client.get_object, container)
        list_objects = functools.partial(client.get_container, container,
                                         limit=limit * 2,
                                         query_string=query_string)
        yield utils._filter_messages(objects, filters, marker, get_object,
                                     list_objects, limit=limit)
        yield marker and marker['next']

    def list(self, queue, project=None, marker=None,
{'fraction_non_alphanumeric': 0.15824343837594496, 'fraction_numerical': 0.0041620657436507265, 'mean_word_length': 1.323662916913361, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4603435', 'n_tokens_mistral': 2653, 'n_tokens_neox': 2493, 'n_words': 838}