<commit_before>"""
kombu.transport.beanstalk
=========================

Beanstalk transport.

:copyright: (c) 2010 - 2013 by <NAME>.
:license: BSD, see LICENSE for more details.

"""
from __future__ import absolute_import, unicode_literals

import socket
import time

from kombu.five import Empty
from kombu.utils.encoding import bytes_to_str
from kombu.utils.json import loads, dumps

from . import virtual

try:
    import beanstalkc
except ImportError:  # pragma: no cover
    beanstalkc = None  # noqa

DEFAULT_PORT = 11300

__author__ = '<NAME> <<EMAIL>>'


class Channel(virtual.Channel):
    _client = None
    _tube_map = {}

    def _format_tube_name(self, tube_name):
        """Valid tube should contain::

            ABCDEFGHIJKLMNOPQRSTUVWXYZ
            abcdefghijklmnopqrstuvwxyz
            0123456789-+/;.$_().

        (see http://bit.ly/1RP3Xjs)

        So the tube_name is generated by `Mailbox` will be invalid.

        This function will change '@' to '.'.

        """
        if '@' not in tube_name:
            new_tube_name = tube_name
        else:
            new_tube_name = tube_name.replace('@', '.')
        self._tube_map[new_tube_name] = tube_name
        return new_tube_name

    def _parse_job(self, job):
        item, dest = None, None
        if job:
            try:
                item = loads(bytes_to_str(job.body))
                dest = job.stats()['tube']
                dest = self._tube_map[dest]
            except Exception:
                job.bury()
            else:
                job.delete()
        else:
            # Sleep for one millisecond in case the queue is empty.
            # This ensures that we don't pound beanstalkd repeatedly
            # because of the timeout=0 that we use when we are reserving jobs.
            time.sleep(0.001) 
            raise Empty()
        return item, dest

    def _put(self, queue, message, **kwargs):
        extra = {}
        priority = message['properties']['delivery_info']['priority']
        ttr = message['properties'].get('ttr')
        if ttr is not None:
            extra['ttr'] = ttr

        self.client.use(queue)
        self.client.put(dumps(message), priority=priority, **extra)

    def _get(self, queue):
        if queue not in self.client.watching():
            queue = self._format_tube_name(queue)
            self.client.watch(queue)

        [self.client.ignore(active) for active in self.client.watching()
         if active != queue]

        # use timeout = 0 so that we don't block on beanstalkd.
        # in _parse_job(), we sleep for a millisecond if we get an empty job.
        # That sleep will allow us not pound beanstalkd because of timeout=0.
        # timeout of 0 is needed because we can't afford to block on beanstalkd 
        # for a good 1 second in case the queue is empty.
        job = self.client.reserve(timeout=0)  
        item, dest = self._parse_job(job)
        return item

    def _get_many(self, queues, timeout=1):
        # timeout of None will cause beanstalk to timeout waiting
        # for a new request
        if timeout is None:
            timeout = 1

        watching = self.client.watching()
        for queue in queues:
            self._format_tube_name(queue)

        [self.client.watch(tube) for tube in self._tube_map
         if tube not in watching]

        [self.client.ignore(active) for active in watching
         if active not in self._tube_map]

        # use timeout = 0 so that we don't block on beanstalkd.
        # in _parse_job(), we sleep for a millisecond if we get an empty job.
        # That sleep will allow us not pound beanstalkd because of timeout=0.
        # timeout of 0 is needed because we can't afford to block on beanstalkd 
        # for a good 1 second in case the queue is empty.
        job = self.client.reserve(timeout=0) <commit_msg>Removed some trailing spaces in transport/beanstalk.py<commit_after>"""
kombu.transport.beanstalk
=========================

Beanstalk transport.

:copyright: (c) 2010 - 2013 by <NAME>.
:license: BSD, see LICENSE for more details.

"""
from __future__ import absolute_import, unicode_literals

import socket
import time

from kombu.five import Empty
from kombu.utils.encoding import bytes_to_str
from kombu.utils.json import loads, dumps

from . import virtual

try:
    import beanstalkc
except ImportError:  # pragma: no cover
    beanstalkc = None  # noqa

DEFAULT_PORT = 11300

__author__ = '<NAME> <<EMAIL>>'


class Channel(virtual.Channel):
    _client = None
    _tube_map = {}

    def _format_tube_name(self, tube_name):
        """Valid tube should contain::

            ABCDEFGHIJKLMNOPQRSTUVWXYZ
            abcdefghijklmnopqrstuvwxyz
            0123456789-+/;.$_().

        (see http://bit.ly/1RP3Xjs)

        So the tube_name is generated by `Mailbox` will be invalid.

        This function will change '@' to '.'.

        """
        if '@' not in tube_name:
            new_tube_name = tube_name
        else:
            new_tube_name = tube_name.replace('@', '.')
        self._tube_map[new_tube_name] = tube_name
        return new_tube_name

    def _parse_job(self, job):
        item, dest = None, None
        if job:
            try:
                item = loads(bytes_to_str(job.body))
                dest = job.stats()['tube']
                dest = self._tube_map[dest]
            except Exception:
                job.bury()
            else:
                job.delete()
        else:
            # Sleep for one millisecond in case the queue is empty.
            # This ensures that we don't pound beanstalkd repeatedly
            # because of the timeout=0 that we use when we are reserving jobs.
            time.sleep(0.001)
            raise Empty()
        return item, dest

    def _put(self, queue, message, **kwargs):
        extra = {}
        priority = message['properties']['delivery_info']['priority']
        ttr = message['properties'].get('ttr')
        if ttr is not None:
            extra['ttr'] = ttr

        self.client.use(queue)
        self.client.put(dumps(message), priority=priority, **extra)

    def _get(self, queue):
        if queue not in self.client.watching():
            queue = self._format_tube_name(queue)
            self.client.watch(queue)

        [self.client.ignore(active) for active in self.client.watching()
         if active != queue]

        # use timeout = 0 so that we don't block on beanstalkd.
        # in _parse_job(), we sleep for a millisecond if we get an empty job.
        # That sleep will allow us not pound beanstalkd because of timeout=0.
        # timeout of 0 is needed because we can't afford to block on beanstalkd
        # for a good 1 second in case the queue is empty.
        job = self.client.reserve(timeout=0)
        item, dest = self._parse_job(job)
        return item

    def _get_many(self, queues, timeout=1):
        # timeout of None will cause beanstalk to timeout waiting
        # for a new request
        if timeout is None:
            timeout = 1

        watching = self.client.watching()
        for queue in queues:
            self._format_tube_name(queue)

        [self.client.watch(tube) for tube in self._tube_map
         if tube not in watching]

        [self.client.ignore(active) for active in watching
         if active not in self._tube_map]

        # use timeout = 0 so that we don't block on beanstalkd.
        # in _parse_job(), we sleep for a millisecond if we get an empty job.
        # That sleep will allow us not pound beanstalkd because of timeout=0.
        # timeout of 0 is needed because we can't afford to block on beanstalkd
        # for a good 1 second in case the queue is empty.
        job = self.client.reserve(timeout=0)
{'fraction_non_alphanumeric': 0.09195106245975532, 'fraction_numerical': 0.01081777205408886, 'mean_word_length': 2.1517857142857144, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2106696', 'n_tokens_mistral': 2373, 'n_tokens_neox': 2192, 'n_words': 757}