<commit_before>from kombu.transport import virtual
from boto.sqs.connection import SQSConnection

try:
    from termcolor import cprint
except ImportError:
    def cprint(s, c):
        print s
    
from kombu.utils import cached_property
from Queue import Empty
from anyjson import serialize, deserialize

def pr(s):
    cprint(s, 'green')        

DEBUG = False

class Channel(virtual.Channel):
    Client = SQSConnection
    
    DOT_REPLECEMENT = '___'
    supports_fanout = False

    def _lookup(self, exchange, routing_key, default="ae.undeliver"):
        """Find all queues matching `routing_key` for the given `exchange`.

        Returns `default` if no queues matched.

        """
        DEBUG and pr('_lookup: %s, %s' % (exchange, routing_key))
        queues = []
        
        #this is a hack for event sending
        #I just send it to all queues  with celeryev
        #In redis backend it looks like same way, but all queues are saved in set
        #but they are saved in _queue_bind witch is called only if supports_fanout == True
        if exchange == 'celeryev':
            for q in self.client.get_all_queues():
                if q.name.split(self.DOT_REPLECEMENT)[0] == 'celeryev':
                    queues.append(q.name)
        
        if queues:
            return queues
        
        return super(Channel, self)._lookup(exchange, routing_key, default)
    
    def _get(self, queue, timeout=None):
        """Get next message from `queue`."""
        DEBUG and pr('>>> Channel._get: %s' % queue)
        q = self._get_queue(queue)
        m = q.read()
        if m:
            q.delete_message(m)
            return deserialize(m.get_body())
        raise Empty()

    def _put(self, queue, message, **kwargs):
        """Put `message` onto `queue`."""
        DEBUG and pr('>>> Channel._put: %s, %s' % (queue, message))
        q = self._get_queue(queue)
        m = q.new_message(serialize(message))
        q.write(m)

    def _purge(self, queue):
        """Remove all messages from `queue`."""
        DEBUG and pr('>>> Channel._purge: %s' % queue)
        return self._get_queue(queue).clear()

    def _size(self, queue):
        """Return the number of messages in `queue` as an :class:`int`."""
        DEBUG and pr('>>> Channel._size: %s' % queue)
        return self._get_queue(queue).count()

    def _delete(self, queue):
        """Delete `queue`.

        This just purges the queue, if you need to do more you can
        override this method.

        """
        DEBUG and pr('>>> Channel._delete: %s' % queue)
        self._purge(queue)
        self._get_queue(queue).delete()

    def _new_queue(self, queue, **kwargs):
        """Create new queue.

        Some implementations needs to do additiona actions when
        the queue is created.  You can do so by overriding this
        method.

        """
        DEBUG and pr('>>> Channel._new_queue: %s' % queue)
        return self._get_queue(queue)

    def _has_queue(self, queue, **kwargs):
        """Verify that queue exists.

        Should return :const:`True` if the queue exists or :const:`False`
        otherwise.

        """
        DEBUG and pr('>>> Channel._has_queue: %s' % queue)
        return True
    
    def _get_queue(self, queue):
        # "." is not valid symbol in queue name for SQS. Maybe this should be more pretty.
        q = self.client.create_queue(queue.replace('.', self.DOT_REPLECEMENT))
        return q
    
    def _poll(self, cycle, timeout=None):
        """Poll a list of queues for available messages."""
        DEBUG and pr('>>> Channel._poll: %s' % cycle)
        return cycle.get()

    def _create_client(self):
        DEBUG and pr('>>> Channel._create_client')
        access_key = self.connection.client.userid
        secret_key = self.connection.client.password           
        return self.Client(access_key, secret_key)

    @cached_property
    def client(self):
        return self._create_client()

    @client.deleter
    def client(self, client):
        client.connection.disconnect()
        
class Transport(virtual.Transport):
    Channel = Channel
    
    def establish_connection(self):<commit_msg>add prefixes for SQS queues<commit_after>from kombu.transport import virtual
from boto.sqs.connection import SQSConnection

try:
    from termcolor import cprint
except ImportError:
    def cprint(s, c):
        print s
    
from kombu.utils import cached_property
from Queue import Empty
from anyjson import serialize, deserialize

def pr(s):
    cprint(s, 'green')        

DEBUG = False

class Channel(virtual.Channel):
    Client = SQSConnection
    
    DOT_REPLECEMENT = '___'
    supports_fanout = False

    def __init__(self, connection, **kwargs):
        self.queue_prefix = connection.client.virtual_host or ''
        super(Channel, self).__init__(connection, **kwargs)

    def _lookup(self, exchange, routing_key, default="ae.undeliver"):
        """Find all queues matching `routing_key` for the given `exchange`.

        Returns `default` if no queues matched.

        """
        DEBUG and pr('_lookup: %s, %s' % (exchange, routing_key))
        queues = []
        
        #this is a hack for event sending
        #I just send it to all queues  with celeryev
        #In redis backend it looks like same way, but all queues are saved in set
        #but they are saved in _queue_bind witch is called only if supports_fanout == True
        if exchange == 'celeryev':
            for q in self.client.get_all_queues():
                parts = q.name.split(self.DOT_REPLECEMENT)
                if parts[0] == self.queue_prefix and parts[1] == 'celeryev':
                    queues.append(self.DOT_REPLECEMENT.join(parts[1:]))
        
        if queues:
            return queues
        
        return super(Channel, self)._lookup(exchange, routing_key, default)
    
    def _get(self, queue, timeout=None):
        """Get next message from `queue`."""
        DEBUG and pr('>>> Channel._get: %s' % queue)
        q = self._get_queue(queue)
        m = q.read()
        if m:
            q.delete_message(m)
            return deserialize(m.get_body())
        raise Empty()

    def _put(self, queue, message, **kwargs):
        """Put `message` onto `queue`."""
        DEBUG and pr('>>> Channel._put: %s, %s' % (queue, message))
        q = self._get_queue(queue)
        m = q.new_message(serialize(message))
        q.write(m)

    def _purge(self, queue):
        """Remove all messages from `queue`."""
        DEBUG and pr('>>> Channel._purge: %s' % queue)
        return self._get_queue(queue).clear()

    def _size(self, queue):
        """Return the number of messages in `queue` as an :class:`int`."""
        DEBUG and pr('>>> Channel._size: %s' % queue)
        return self._get_queue(queue).count()

    def _delete(self, queue):
        """Delete `queue`.

        This just purges the queue, if you need to do more you can
        override this method.

        """
        DEBUG and pr('>>> Channel._delete: %s' % queue)
        print pr('>>> Channel._delete: %s' % queue)
        self._purge(queue)
        self._get_queue(queue).delete()

    def _new_queue(self, queue, **kwargs):
        """Create new queue.

        Some implementations needs to do additiona actions when
        the queue is created.  You can do so by overriding this
        method.

        """
        DEBUG and pr('>>> Channel._new_queue: %s' % queue)
        return self._get_queue(queue)

    def _has_queue(self, queue, **kwargs):
        """Verify that queue exists.

        Should return :const:`True` if the queue exists or :const:`False`
        otherwise.

        """
        DEBUG and pr('>>> Channel._has_queue: %s' % queue)
        return True
    
    def _get_queue(self, queue):
        # "." is not valid symbol in queue name for SQS. Maybe this should be more pretty.
        queue = self.queue_prefix+'.'+queue
        q = self.client.create_queue(queue.replace('.', self.DOT_REPLECEMENT))
        return q
    
    def _poll(self, cycle, timeout=None):
        """Poll a list of queues for available messages."""
        DEBUG and pr('>>> Channel._poll: %s' % cycle)
        return cycle.get()

    def _create_client(self):
        DEBUG and pr('>>> Channel._create_client')
        access_key = self.connection.client.userid
        secret_key = self.connection.client.password           
        return self.Client(access_key, secret_key)

    @cached_property
    def client(self):
        return self._create_client()

    @client.deleter
    def client(self, client):
        client.connection.disconnect()
        
class Transport(virtual.Transport):
    Channel = Channel
    
    def establish_connection(self):
{'fraction_non_alphanumeric': 0.11268571428571429, 'fraction_numerical': 0.00045714285714285713, 'mean_word_length': 2.3185437997724687, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 60, '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': '1057463', 'n_tokens_mistral': 2586, 'n_tokens_neox': 2517, 'n_words': 795}