<commit_before>"""AMQPStorm Connection.IO."""

import logging
import select
import socket
import threading
from errno import EAGAIN
from errno import EINTR
from errno import EWOULDBLOCK
from time import sleep

from amqpstorm import compatibility
from amqpstorm.base import FRAME_MAX
from amqpstorm.base import IDLE_WAIT
from amqpstorm.compatibility import ssl
from amqpstorm.exception import AMQPConnectionError

EMPTY_BUFFER = bytes()
LOGGER = logging.getLogger(__name__)


class Poller(object):
    """Socket Read Poller."""

    def __init__(self, fileno, exceptions, timeout=5):
        self._fileno = fileno
        self._exceptions = exceptions
        self.timeout = timeout

    @property
    def fileno(self):
        """Socket Fileno.

        :return:
        """
        return self._fileno

    @property
    def is_ready(self):
        """Is Socket Ready.

        :rtype: tuple
        """
        try:
            ready, _, _ = select.select([self.fileno], [], [],
                                        self.timeout)
            return bool(ready)
        except select.error as why:
            if why.args[0] != EINTR:
                self._exceptions.append(AMQPConnectionError(why))
        return False


class IO(object):
    """Internal Input/Output handler."""

    def __init__(self, parameters, exceptions=None, on_read=None):
        self._exceptions = exceptions
        self._lock = threading.Lock()
        self._inbound_thread = None
        self._on_read = on_read
        self._running = threading.Event()
        self._parameters = parameters
        self.data_in = EMPTY_BUFFER
        self.poller = None
        self.socket = None
        self.use_ssl = self._parameters['ssl']

    def close(self):
        """Close Socket.

        :return:
        """
        self._lock.acquire()
        try:
            self._running.clear()
            if self._inbound_thread:
                self._inbound_thread.join(timeout=self._parameters['timeout'])
            self._inbound_thread = None
            self.poller = None
            if self.socket:
                self.socket.close()
            self.socket = None
        finally:
            self._lock.release()

    def open(self):
        """Open Socket and establish a connection.
<commit_msg>Re-worked Connection.close to be more responsive<commit_after>"""AMQPStorm Connection.IO."""

import logging
import select
import socket
import threading
from errno import EAGAIN
from errno import EINTR
from errno import EWOULDBLOCK
from time import sleep

from amqpstorm import compatibility
from amqpstorm.base import FRAME_MAX
from amqpstorm.base import IDLE_WAIT
from amqpstorm.compatibility import ssl
from amqpstorm.exception import AMQPConnectionError

EMPTY_BUFFER = bytes()
LOGGER = logging.getLogger(__name__)


class Poller(object):
    """Socket Read Poller."""

    def __init__(self, fileno, exceptions, timeout=5):
        self._fileno = fileno
        self._exceptions = exceptions
        self.timeout = timeout

    @property
    def fileno(self):
        """Socket Fileno.

        :return:
        """
        return self._fileno

    @property
    def is_ready(self):
        """Is Socket Ready.

        :rtype: tuple
        """
        try:
            ready, _, _ = select.select([self.fileno], [], [],
                                        self.timeout)
            return bool(ready)
        except select.error as why:
            if why.args[0] != EINTR:
                self._exceptions.append(AMQPConnectionError(why))
        return False


class IO(object):
    """Internal Input/Output handler."""

    def __init__(self, parameters, exceptions=None, on_read=None):
        self._exceptions = exceptions
        self._lock = threading.Lock()
        self._inbound_thread = None
        self._on_read = on_read
        self._running = threading.Event()
        self._parameters = parameters
        self.data_in = EMPTY_BUFFER
        self.poller = None
        self.socket = None
        self.use_ssl = self._parameters['ssl']

    def close(self):
        """Close Socket.

        :return:
        """
        self._lock.acquire()
        try:
            self._running.clear()
            if self.socket:
                self.socket.close()
            if self._inbound_thread:
                self._inbound_thread.join(timeout=self._parameters['timeout'])
            self._inbound_thread = None
            self.poller = None
            self.socket = None
        finally:
            self._lock.release()

    def open(self):
        """Open Socket and establish a connection.

{'fraction_non_alphanumeric': 0.09294245385450597, 'fraction_numerical': 0.0008686210640608035, 'mean_word_length': 2.0829986613119145, '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': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1494956', 'n_tokens_mistral': 1278, 'n_tokens_neox': 1229, 'n_words': 321}