<commit_before>import re
import logging
from collections import defaultdict
from contextlib import contextmanager

import mock
import kombu
import pkg_resources

log = logging.getLogger(__name__)

class Connection(object):

    def __init__(self, hostname, port, userid, password, vhost):
        self._conn = kombu.BrokerConnection(
            hostname=hostname,
            port=port,
            userid=userid,
            password=password,
            virtual_host=vhost)
        self._channel_pool = self._conn.ChannelPool(preload=2, limit=None)
        self._exchanges = dict(
            audit=kombu.Exchange('audit'),
            react=kombu.Exchange('react'))

    @contextmanager
    def channel(self):
        ch = self._channel_pool.acquire()
        try:
            yield ch
        finally:
            if ch.is_open: ch.release()
            else: log.info('ch is not open, not returning to pool')

    @contextmanager
    def producer(self, xn):<commit_msg>[#1601] Handle channel pool exhausted errors by creating a new connection & pool<commit_after>import re
import logging
from collections import defaultdict
from contextlib import contextmanager

import mock
import kombu
import pkg_resources

log = logging.getLogger(__name__)

class Connection(object):

    def __init__(self, hostname, port, userid, password, vhost):
        self._conn_proto = kombu.BrokerConnection(
            hostname=hostname,
            port=port,
            userid=userid,
            password=password,
            virtual_host=vhost)
        self._connection_pool = self._conn_proto.Pool(preload=1, limit=None)
        self._exchanges = dict(
            audit=kombu.Exchange('audit'),
            react=kombu.Exchange('react'))
        self.reset()

    def reset(self):
        self._conn = self._connection_pool.acquire()
        self._channel_pool = self._conn.ChannelPool(preload=2, limit=10)

    @contextmanager
    def channel(self):
        try:
            ch = self._channel_pool.acquire()
        except kombu.exceptions.ChannelLimitExceeded:
            log.info('Channel pool exhausted, opening a new connection')
            self.reset()
            ch = self._channel_pool.acquire()
        try:
            yield ch
        finally:
            if ch.is_open: ch.release()
            else: log.info('ch is not open, not returning to pool')

    @contextmanager
    def producer(self, xn):
{'fraction_non_alphanumeric': 0.0891746163417669, 'fraction_numerical': 0.0037328909166321027, 'mean_word_length': 2.470503597122302, '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': '1277813', 'n_tokens_mistral': 682, 'n_tokens_neox': 669, 'n_words': 167}