<commit_before>"""
Channel0 is used for connection level communication between RabbitMQ and the
client on channel 0.

"""
import locale
import logging
try:
    import queue
except ImportError:
    import Queue as queue
import sys
import threading

from pamqp import header
from pamqp import specification

from rabbitpy import __version__
from rabbitpy import base
from rabbitpy import events

LOGGER = logging.getLogger(__name__)


class Channel0(threading.Thread, base.AMQPChannel):
    """Channel0 is used to negotiate a connection with RabbitMQ and for
    processing and dispatching events on channel 0 once connected.

    :param dict connection_args: Data required to negotiate the connection
    :param Queue.Queue inbound: Inbound frame queue
    :param Queue.Queue outbound: Outbound frame queue

    """
    CHANNEL = 0
    DEFAULT_CLOSE_CODE = 200
    DEFAULT_CLOSE_REASON = 'Normal Shutdown'
    DEFAULT_LOCALE = 'en-US'

    def __init__(self, group=None, target=None, name=None, args=(), kwargs={}):
        super(Channel0, self).__init__(group, target, name, args, kwargs)
        self._channel_id = 0
        self._args = kwargs['connection_args']
        self._events = kwargs['events']
        self._exceptions = kwargs['exceptions']
        self._read_queue = kwargs['inbound']
        self._write_queue = kwargs['outbound']
        self._heartbeat = 0
        self._maximum_channels = 0
        self._state = self.CLOSED
        self.maximum_frame_size = specification.FRAME_MAX_SIZE
        self.minimum_frame_size = specification.FRAME_MIN_SIZE

    def close(self):
        if self.open:
            self._set_state(self.CLOSING)
            self._write_frame(self._build_close_frame())
            self._events.set(events.CHANNEL0_CLOSED)
            self._set_state(self.CLOSED)
        else:
            LOGGER.info('Bypassing sending Connection.Close, already closed')

    @property
    def maximum_channels(self):
        return self._maximum_channels

    def run(self):
        try:
            self._run()
        except Exception as exception:
            self._exceptions.put(exception)
            if self.open:
                self.close()

    def _run(self):
        self._set_state(self.OPENING)
        self._write_protocol_header()

        # If we can negotiate a connection, do so
        if self._connection_start():
            self._connection_tune()
            self._connection_open()

            # Loop as long as the connection is open, waiting for RPC requests
            while self.open and not self._events.is_set(events.CHANNEL0_CLOSE):
                if not self._exceptions.empty():
                    LOGGER.debug('exiting main loop due to exceptions')<commit_msg>Check for the event instead of the queue size<commit_after>"""
Channel0 is used for connection level communication between RabbitMQ and the
client on channel 0.

"""
import locale
import logging
try:
    import queue
except ImportError:
    import Queue as queue
import sys
import threading

from pamqp import header
from pamqp import specification

from rabbitpy import __version__
from rabbitpy import base
from rabbitpy import events

LOGGER = logging.getLogger(__name__)


class Channel0(threading.Thread, base.AMQPChannel):
    """Channel0 is used to negotiate a connection with RabbitMQ and for
    processing and dispatching events on channel 0 once connected.

    :param dict connection_args: Data required to negotiate the connection
    :param Queue.Queue inbound: Inbound frame queue
    :param Queue.Queue outbound: Outbound frame queue

    """
    CHANNEL = 0
    DEFAULT_CLOSE_CODE = 200
    DEFAULT_CLOSE_REASON = 'Normal Shutdown'
    DEFAULT_LOCALE = 'en-US'

    def __init__(self, group=None, target=None, name=None, args=(), kwargs={}):
        super(Channel0, self).__init__(group, target, name, args, kwargs)
        self._channel_id = 0
        self._args = kwargs['connection_args']
        self._events = kwargs['events']
        self._exceptions = kwargs['exceptions']
        self._read_queue = kwargs['inbound']
        self._write_queue = kwargs['outbound']
        self._heartbeat = 0
        self._maximum_channels = 0
        self._state = self.CLOSED
        self.maximum_frame_size = specification.FRAME_MAX_SIZE
        self.minimum_frame_size = specification.FRAME_MIN_SIZE

    def close(self):
        if self.open:
            self._set_state(self.CLOSING)
            self._write_frame(self._build_close_frame())
            self._events.set(events.CHANNEL0_CLOSED)
            self._set_state(self.CLOSED)
        else:
            LOGGER.info('Bypassing sending Connection.Close, already closed')

    @property
    def maximum_channels(self):
        return self._maximum_channels

    def run(self):
        try:
            self._run()
        except Exception as exception:
            self._exceptions.put(exception)
            if self.open:
                self.close()
            self._events.set(events.EXCEPTION_RAISED)

    def _run(self):
        self._set_state(self.OPENING)
        self._write_protocol_header()

        # If we can negotiate a connection, do so
        if self._connection_start():
            self._connection_tune()
            self._connection_open()

            # Loop as long as the connection is open, waiting for RPC requests
            while self.open and not self._events.is_set(events.CHANNEL0_CLOSE):
                if self._events.is_set(events.EXCEPTION_RAISED):
                    LOGGER.debug('exiting main loop due to exceptions')
{'fraction_non_alphanumeric': 0.07778177889809147, 'fraction_numerical': 0.005401512423478574, 'mean_word_length': 2.691029900332226, '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': '73694', 'n_tokens_mistral': 1575, 'n_tokens_neox': 1512, 'n_words': 439}