<commit_before>from __future__ import absolute_import

from contextlib import contextmanager

import logging
import gevent
import gevent.pool
import kombu
import kombu.mixins
import kombu.pools

from lymph.events.base import BaseEventSystem
from lymph.core.events import Event


logger = logging.getLogger(__name__)


DEFAULT_SERIALIZER = 'lymph-msgpack'
DEFAULT_EXCHANGE = 'lymph'


class EventConsumer(kombu.mixins.ConsumerMixin):
    def __init__(self, connection, queue, handler, spawn=gevent.spawn):
        self.connection = connection
        self.queue = queue
        self.handler = handler
        self.spawn = spawn
        self.greenlet = None

    def get_consumers(self, Consumer, channel):
        return [Consumer(queues=[self.queue], callbacks=[self.on_kombu_message])]

    def create_connection(self):
        return kombu.pools.connections[self.connection].acquire(block=True)

    def on_kombu_message(self, body, message):
        logger.debug("received kombu message %r", body)

        def message_handler():
            try:
                event = Event.deserialize(body)
                self.handler(event)
                message.ack()
            except:
                logger.exception('failed to handle event from queue %r', self.handler.queue_name)

        if self.handler.sequential:
            message_handler()
        else:
            self.spawn(message_handler)

    def start(self):
        if self.greenlet:
            return
        self.should_stop = False
        self.greenlet = self.spawn(self.run)

    def stop(self):
        if not self.greenlet:
            return
        self.should_stop = True
        self.greenlet.join()
        self.greenlet = None


class KombuEventSystem(BaseEventSystem):
    def __init__(self, connection, exchange_name, serializer=DEFAULT_SERIALIZER):
        self.connection = connection
        self.exchange = kombu.Exchange(exchange_name, 'topic', durable=True)
        self.greenlets = gevent.pool.Group()
        self.serializer = serializer
        self.consumers_by_queue = {}

    def on_stop(self):<commit_msg>Reject kombu messages on error<commit_after>from __future__ import absolute_import

from contextlib import contextmanager

import logging
import gevent
import gevent.pool
import kombu
import kombu.mixins
import kombu.pools

from lymph.events.base import BaseEventSystem
from lymph.core.events import Event


logger = logging.getLogger(__name__)


DEFAULT_SERIALIZER = 'lymph-msgpack'
DEFAULT_EXCHANGE = 'lymph'


class EventConsumer(kombu.mixins.ConsumerMixin):
    def __init__(self, connection, queue, handler, spawn=gevent.spawn):
        self.connection = connection
        self.queue = queue
        self.handler = handler
        self.spawn = spawn
        self.greenlet = None

    def get_consumers(self, Consumer, channel):
        return [Consumer(queues=[self.queue], callbacks=[self.on_kombu_message])]

    def create_connection(self):
        return kombu.pools.connections[self.connection].acquire(block=True)

    def on_kombu_message(self, body, message):
        logger.debug("received kombu message %r", body)

        def message_handler():
            try:
                event = Event.deserialize(body)
                self.handler(event)
                message.ack()
            except:
                logger.exception('failed to handle event from queue %r', self.handler.queue_name)
                # FIXME: add requeue support here
                message.reject()

        if self.handler.sequential:
            message_handler()
        else:
            self.spawn(message_handler)

    def start(self):
        if self.greenlet:
            return
        self.should_stop = False
        self.greenlet = self.spawn(self.run)

    def stop(self):
        if not self.greenlet:
            return
        self.should_stop = True
        self.greenlet.join()
        self.greenlet = None


class KombuEventSystem(BaseEventSystem):
    def __init__(self, connection, exchange_name, serializer=DEFAULT_SERIALIZER):
        self.connection = connection
        self.exchange = kombu.Exchange(exchange_name, 'topic', durable=True)
        self.greenlets = gevent.pool.Group()
        self.serializer = serializer
        self.consumers_by_queue = {}

    def on_stop(self):
{'fraction_non_alphanumeric': 0.08623895862389586, 'fraction_numerical': 0.0, 'mean_word_length': 2.612930310663308, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '5964347', 'n_tokens_mistral': 1264, 'n_tokens_neox': 1207, 'n_words': 275}