<commit_before>"""AMQP-Storm Connection.Channel0."""

import locale
import logging
import platform

from pamqp import specification as pamqp_spec
from pamqp.heartbeat import Heartbeat
from pamqp.specification import Connection as pamqp_connection

from amqpstorm import __version__
from amqpstorm.base import FRAME_MAX
from amqpstorm.base import Stateful
from amqpstorm.exception import AMQPConnectionError
from amqpstorm.compatibility import try_utf8_decode

AUTH_MECHANISM = 'PLAIN'
LOCALE = locale.getdefaultlocale()[0] or 'en_US'
LOGGER = logging.getLogger(__name__)


class Channel0(object):
    """Connection.Channel0."""

    def __init__(self, connection):
        super(Channel0, self).__init__()
        self.is_blocked = False
        self.server_properties = {}
        self.parameters = connection.parameters
        self._connection = connection
        self._heartbeat = self.parameters['heartbeat']

    def on_frame(self, frame_in):
        """Handle frame sent to channel 0.

        :param frame_in: Amqp frame.
        :return:
        """
        LOGGER.debug('Frame Received: %s', frame_in.name)
        if frame_in.name == 'Heartbeat':
            self._connection.heartbeat.register_heartbeat()
            self._write_frame(Heartbeat())
        elif frame_in.name == 'Connection.Start':
            self.server_properties = frame_in.server_properties
            self._send_start_ok_frame(frame_in)
        elif frame_in.name == 'Connection.Tune':
            self._send_tune_ok_frame()
            self._send_open_connection()
        elif frame_in.name == 'Connection.OpenOk':
            self._set_connection_state(Stateful.OPEN)
        elif frame_in.name == 'Connection.Close':
            self._close_connection(frame_in)
        elif frame_in.name == 'Connection.CloseOk':
            self._close_connection_ok()
        elif frame_in.name == 'Connection.Blocked':
            self.is_blocked = True
            LOGGER.warning('Connection is blocked by remote server: %s',
                           try_utf8_decode(frame_in.reason))
        elif frame_in.name == 'Connection.Unblocked':
            self.is_blocked = False
            LOGGER.info('Connection is no longer blocked by remote server')
        else:
            LOGGER.error('[Channel0] Unhandled Frame: %s', frame_in.name)

    def send_close_connection_frame(self):
        """Send Connection Close frame.

        :return:
        """
        self._write_frame(pamqp_spec.Connection.Close())

    def _close_connection(self, frame_in):
        """Connection Close.

        :param pamqp_spec.Connection.Close frame_in: Amqp frame.
        :return:
        """
        self._set_connection_state(Stateful.CLOSED)
        if frame_in.reply_code != 200:
            reply_text = try_utf8_decode(frame_in.reply_text)
            message = ('Connection was closed by remote server: %s'
                       % reply_text)
            exception = AMQPConnectionError(message,
                                            reply_code=frame_in.reply_code)
            self._connection.exceptions.append(exception)

    def _close_connection_ok(self):
        """Connection CloseOk.

        :return:
        """
        self._set_connection_state(Stateful.CLOSED)

    def _set_connection_state(self, state):
        """Set Connection state.

        :param state:
        :return:
        """
        self._connection.set_state(state)

    def _write_frame(self, frame_out):
        """Write a pamqp frame from channel0.

        :param frame_out: Amqp frame.
        :return:
        """
        self._connection.write_frame(0, frame_out)

    def _send_start_ok_frame(self, frame_in):
        """Send Start OK frame.

        :param pamqp_spec.Frame frame_out: Amqp frame.
        :return:
        """
        if 'PLAIN' not in try_utf8_decode(frame_in.mechanisms):
            exception = AMQPConnectionError('Unsupported Security Mechanism(s)'
                                            ': %s' % frame_in.mechanisms)
            self._connection.exceptions.append(exception)
            return
        credentials = self._plain_credentials()
        frame = pamqp_connection.StartOk(
            mechanism=AUTH_MECHANISM,
            client_properties=self._client_properties(),
            response=credentials,
            locale=LOCALE)
        self._write_frame(frame)

    def _send_tune_ok_frame(self):
        """Send Tune OK frame.

        :return:
        """
        frame = pamqp_connection.TuneOk(channel_max=0,
                                        frame_max=FRAME_MAX,
                                        heartbeat=self._heartbeat)
        self._write_frame(frame)

    def _send_open_connection(self):
        """Send Open Connection frame.

        :return:
        """
        frame = pamqp_connection.Open(
            virtual_host=self.parameters['virtual_host']
        )
        self._write_frame(frame)

    def _plain_credentials(self):
        """AMQP Plain Credentials.

        :rtype: str
        """
        return '\0%s\0%s' % (self.parameters['username'],
                             self.parameters['password'])

    @staticmethod
    def _client_properties():
        """AMQP Library Properties.

        :rtype: dict
        """
        return {
            'product': 'AMQP-Storm',
            'platform': 'Python %s' % platform.python_version(),
            'capabilities': {
                'basic.nack': True,
                'connection.blocked': True,
                'publisher_confirms': True,
                'consumer_cancel_notify': True,
                'authentication_failure_close': True,
            },
            'information': 'See https://github.com/eandersson/amqp-storm',
            'version': __version__
        }
<commit_msg>Fixing OpenAMQ Support<commit_after>"""AMQP-Storm Connection.Channel0."""

import locale
import logging
import platform

from pamqp import specification as pamqp_spec
from pamqp.heartbeat import Heartbeat
from pamqp.specification import Connection as pamqp_connection

from amqpstorm import __version__
from amqpstorm.base import FRAME_MAX
from amqpstorm.base import Stateful
from amqpstorm.exception import AMQPConnectionError
from amqpstorm.compatibility import try_utf8_decode

AUTH_MECHANISM = 'PLAIN'
LOCALE = locale.getdefaultlocale()[0] or 'en_US'
LOGGER = logging.getLogger(__name__)
MAX_CHANNELS = 65535


class Channel0(object):
    """Connection.Channel0."""

    def __init__(self, connection):
        super(Channel0, self).__init__()
        self.is_blocked = False
        self.server_properties = {}
        self.parameters = connection.parameters
        self._connection = connection
        self._heartbeat = self.parameters['heartbeat']

    def on_frame(self, frame_in):
        """Handle frame sent to channel 0.

        :param frame_in: Amqp frame.
        :return:
        """
        LOGGER.debug('Frame Received: %s', frame_in.name)
        if frame_in.name == 'Heartbeat':
            self._connection.heartbeat.register_heartbeat()
            self._write_frame(Heartbeat())
        elif frame_in.name == 'Connection.Start':
            self.server_properties = frame_in.server_properties
            self._send_start_ok_frame(frame_in)
        elif frame_in.name == 'Connection.Tune':
            self._send_tune_ok_frame()
            self._send_open_connection()
        elif frame_in.name == 'Connection.OpenOk':
            self._set_connection_state(Stateful.OPEN)
        elif frame_in.name == 'Connection.Close':
            self._close_connection(frame_in)
        elif frame_in.name == 'Connection.CloseOk':
            self._close_connection_ok()
        elif frame_in.name == 'Connection.Blocked':
            self.is_blocked = True
            LOGGER.warning('Connection is blocked by remote server: %s',
                           try_utf8_decode(frame_in.reason))
        elif frame_in.name == 'Connection.Unblocked':
            self.is_blocked = False
            LOGGER.info('Connection is no longer blocked by remote server')
        else:
            LOGGER.error('[Channel0] Unhandled Frame: %s', frame_in.name)

    def send_close_connection_frame(self):
        """Send Connection Close frame.

        :return:
        """
        self._write_frame(pamqp_spec.Connection.Close())

    def _close_connection(self, frame_in):
        """Connection Close.

        :param pamqp_spec.Connection.Close frame_in: Amqp frame.
        :return:
        """
        self._set_connection_state(Stateful.CLOSED)
        if frame_in.reply_code != 200:
            reply_text = try_utf8_decode(frame_in.reply_text)
            message = ('Connection was closed by remote server: %s'
                       % reply_text)
            exception = AMQPConnectionError(message,
                                            reply_code=frame_in.reply_code)
            self._connection.exceptions.append(exception)

    def _close_connection_ok(self):
        """Connection CloseOk.

        :return:
        """
        self._set_connection_state(Stateful.CLOSED)

    def _set_connection_state(self, state):
        """Set Connection state.

        :param state:
        :return:
        """
        self._connection.set_state(state)

    def _write_frame(self, frame_out):
        """Write a pamqp frame from channel0.

        :param frame_out: Amqp frame.
        :return:
        """
        self._connection.write_frame(0, frame_out)

    def _send_start_ok_frame(self, frame_in):
        """Send Start OK frame.

        :param pamqp_spec.Frame frame_out: Amqp frame.
        :return:
        """
        if 'PLAIN' not in try_utf8_decode(frame_in.mechanisms):
            exception = AMQPConnectionError('Unsupported Security Mechanism(s)'
                                            ': %s' % frame_in.mechanisms)
            self._connection.exceptions.append(exception)
            return
        credentials = self._plain_credentials()
        frame = pamqp_connection.StartOk(
            mechanism=AUTH_MECHANISM,
            client_properties=self._client_properties(),
            response=credentials,
            locale=LOCALE)
        self._write_frame(frame)

    def _send_tune_ok_frame(self):
        """Send Tune OK frame.

        :return:
        """
        frame = pamqp_connection.TuneOk(channel_max=MAX_CHANNELS,
                                        frame_max=FRAME_MAX,
                                        heartbeat=self._heartbeat)
        self._write_frame(frame)

    def _send_open_connection(self):
        """Send Open Connection frame.

        :return:
        """
        frame = pamqp_connection.Open(
            virtual_host=self.parameters['virtual_host']
        )
        self._write_frame(frame)

    def _plain_credentials(self):
        """AMQP Plain Credentials.

        :rtype: str
        """
        return '\0%s\0%s' % (self.parameters['username'],
                             self.parameters['password'])

    @staticmethod
    def _client_properties():
        """AMQP Library Properties.

        :rtype: dict
        """
        return {
            'product': 'AMQP-Storm',
            'platform': 'Python %s' % platform.python_version(),
            'capabilities': {
                'basic.nack': True,
                'connection.blocked': True,
                'publisher_confirms': True,
                'consumer_cancel_notify': True,
                'authentication_failure_close': True,
            },
            'information': 'See https://github.com/eandersson/amqp-storm',
            'version': __version__
        }

{'fraction_non_alphanumeric': 0.09666867521735388, 'fraction_numerical': 0.0036153912369802873, 'mean_word_length': 2.1164163090128754, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4472461', 'n_tokens_mistral': 3251, 'n_tokens_neox': 3166, 'n_words': 638}