<commit_before># ***** BEGIN LICENSE BLOCK *****
#
# For copyright and licensing please refer to COPYING.
#
# ***** END LICENSE BLOCK *****

"""
Pika provides multiple adapters to connect to RabbitMQ:

- adapters.select_connection.SelectConnection: A native event based connection
  adapter that implements select, kqueue, poll and epoll.
- adapters.asyncore_connection.AsyncoreConnection: Legacy adapter kept for
  convenience of previous Pika users. It is recommended to use the
  SelectConnection instead of AsyncoreConnection.
- adapters.tornado_connection.TornadoConnection: Connection adapter for use
  with the Tornado web framework.
- adapters.blocking_connection.BlockingConnection: Enables blocking,
  synchronous operation on top of library for simple uses. This is not
  recommended and is included for legacy reasons only.
"""

import errno
import socket
from time import sleep, time

# See if we have SSL support
try:
    import ssl
    SSL = True
except ImportError:
    SSL = False

from pika.connection import Connection, CONNECTION_PROTOCOL, CONNECTION_START,\
    CONNECTION_TUNE
from pika.exceptions import AMQPConnectionError, IncompatibleProtocolError, \
    ProbableAuthenticationError, ProbableAccessDeniedError
import pika.log as log

# Use epoll's constants to keep life easy
READ = 0x0001
WRITE = 0x0004
ERROR = 0x0008

# Connection timeout (2 seconds to open socket)
CONNECTION_TIMEOUT = 2
ERRORS_TO_IGNORE =  [errno.EWOULDBLOCK, errno.EAGAIN, errno.EINTR]


class BaseConnection(Connection):

    def __init__(self, parameters=None,
                       on_open_callback=None,
                       reconnection_strategy=None):

        # Let the developer know we could not import SSL
        if parameters.ssl and not SSL:
            raise Exception("SSL specified but it is not available")

        # Call our parent's __init__
        Connection.__init__(self, parameters, on_open_callback,
                            reconnection_strategy)

    def _init_connection_state(self):
        Connection._init_connection_state(self)

        # Set our defaults
        self.fd = None
        self.ioloop = None

        # Event states (base and current)
        self.base_events = READ | ERROR
        self.event_state = self.base_events
        self.socket = None
        self.write_buffer = None
        self._ssl_connecting = False
        self._ssl_handshake = False

    def _socket_connect(self):
        """Create socket and connect to it, using SSL if enabled"""
        # Create our socket and set our socket options
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
        self.socket.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)

        # Wrap the SSL socket if we SSL turned on
        ssl_text = ""
        if self.parameters.ssl:
            ssl_text = " with SSL"
            if self.parameters.ssl_options:
                # Always overwrite this value
                self.parameters.ssl_options['do_handshake_on_connect'] = \
                    self._ssl_handshake
                self.socket = ssl.wrap_socket(self.socket,
                                              **self.parameters.ssl_options)
            else:
                self.socket = ssl.wrap_socket(self.socket,
                                              do_handshake_on_connect= \
                                                  self._ssl_handshake)

            # Flags for SSL handshake negotiation
            self._ssl_connecting = True

        # Try and connect
        log.info("Connecting fd %d to %s:%i%s", self.socket.fileno(),
                 self.parameters.host,
                 self.parameters.port, ssl_text)
        self.socket.settimeout(CONNECTION_TIMEOUT)
        self.socket.connect((self.parameters.host,
                             self.parameters.port))

        # Set the socket to non-blocking
        self.socket.setblocking(0)

    def _adapter_connect(self):
        """
        Base connection function to be extended as needed
        """

        # Set our remaining attempts to the initial value
        from sys import maxint
        remaining_attempts = self.parameters.connection_attempts or maxint

        # Loop while we have remaining attempts
        while remaining_attempts:
            remaining_attempts -= 1
            try:
                return self._socket_connect()
            except socket.timeout, timeout:
                reason = "timeout"
            except socket.error, err:
                reason = err[-1]
                self.socket.close()

            retry = ''
            if remaining_attempts:
                retry = "Retrying in %i seconds with %i \
retry(s) left" % (self.parameters.retry_delay, remaining_attempts)

            log.warning("Could not connect: %s. %s", reason, retry)

            if remaining_attempts:
                sleep(self.parameters.retry_delay)

        # Log the errors and raise the  exception
        log.error("Could not connect: %s", err[-1])
        raise AMQPConnectionError(err[-1])

    def add_timeout(self, delay_sec, callback):
        deadline = time() + delay_sec
        return self.ioloop.add_timeout(deadline, callback)

    def remove_timeout(self, timeout_id):
        self.ioloop.remove_timeout(timeout_id)

    def _erase_credentials(self):
        pass

    def _flush_outbound(self):
        """
        Call the state manager who will figure out that we need to write.
        """
        self._manage_event_state()
<commit_msg>Fixed small bug in error code handling for connection attempts.<commit_after># ***** BEGIN LICENSE BLOCK *****
#
# For copyright and licensing please refer to COPYING.
#
# ***** END LICENSE BLOCK *****

"""
Pika provides multiple adapters to connect to RabbitMQ:

- adapters.select_connection.SelectConnection: A native event based connection
  adapter that implements select, kqueue, poll and epoll.
- adapters.asyncore_connection.AsyncoreConnection: Legacy adapter kept for
  convenience of previous Pika users. It is recommended to use the
  SelectConnection instead of AsyncoreConnection.
- adapters.tornado_connection.TornadoConnection: Connection adapter for use
  with the Tornado web framework.
- adapters.blocking_connection.BlockingConnection: Enables blocking,
  synchronous operation on top of library for simple uses. This is not
  recommended and is included for legacy reasons only.
"""

import errno
import socket
from time import sleep, time

# See if we have SSL support
try:
    import ssl
    SSL = True
except ImportError:
    SSL = False

from pika.connection import Connection, CONNECTION_PROTOCOL, CONNECTION_START,\
    CONNECTION_TUNE
from pika.exceptions import AMQPConnectionError, IncompatibleProtocolError, \
    ProbableAuthenticationError, ProbableAccessDeniedError
import pika.log as log

# Use epoll's constants to keep life easy
READ = 0x0001
WRITE = 0x0004
ERROR = 0x0008

# Connection timeout (2 seconds to open socket)
CONNECTION_TIMEOUT = 2
ERRORS_TO_IGNORE =  [errno.EWOULDBLOCK, errno.EAGAIN, errno.EINTR]


class BaseConnection(Connection):

    def __init__(self, parameters=None,
                       on_open_callback=None,
                       reconnection_strategy=None):

        # Let the developer know we could not import SSL
        if parameters.ssl and not SSL:
            raise Exception("SSL specified but it is not available")

        # Call our parent's __init__
        Connection.__init__(self, parameters, on_open_callback,
                            reconnection_strategy)

    def _init_connection_state(self):
        Connection._init_connection_state(self)

        # Set our defaults
        self.fd = None
        self.ioloop = None

        # Event states (base and current)
        self.base_events = READ | ERROR
        self.event_state = self.base_events
        self.socket = None
        self.write_buffer = None
        self._ssl_connecting = False
        self._ssl_handshake = False

    def _socket_connect(self):
        """Create socket and connect to it, using SSL if enabled"""
        # Create our socket and set our socket options
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
        self.socket.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)

        # Wrap the SSL socket if we SSL turned on
        ssl_text = ""
        if self.parameters.ssl:
            ssl_text = " with SSL"
            if self.parameters.ssl_options:
                # Always overwrite this value
                self.parameters.ssl_options['do_handshake_on_connect'] = \
                    self._ssl_handshake
                self.socket = ssl.wrap_socket(self.socket,
                                              **self.parameters.ssl_options)
            else:
                self.socket = ssl.wrap_socket(self.socket,
                                              do_handshake_on_connect= \
                                                  self._ssl_handshake)

            # Flags for SSL handshake negotiation
            self._ssl_connecting = True

        # Try and connect
        log.info("Connecting fd %d to %s:%i%s", self.socket.fileno(),
                 self.parameters.host,
                 self.parameters.port, ssl_text)
        self.socket.settimeout(CONNECTION_TIMEOUT)
        self.socket.connect((self.parameters.host,
                             self.parameters.port))

        # Set the socket to non-blocking
        self.socket.setblocking(0)

    def _adapter_connect(self):
        """
        Base connection function to be extended as needed
        """

        # Set our remaining attempts to the initial value
        from sys import maxint
        remaining_attempts = self.parameters.connection_attempts or maxint

        # Loop while we have remaining attempts
        while remaining_attempts:
            remaining_attempts -= 1
            try:
                return self._socket_connect()
            except socket.timeout, timeout:
                reason = "timeout"
            except socket.error, err:
                reason = err[-1]
                self.socket.close()

            retry = ''
            if remaining_attempts:
                retry = "Retrying in %i seconds with %i \
retry(s) left" % (self.parameters.retry_delay, remaining_attempts)

            log.warning("Could not connect: %s. %s", reason, retry)

            if remaining_attempts:
                sleep(self.parameters.retry_delay)

        # Log the errors and raise the  exception
        log.error("Could not connect: %s", reason)
        raise AMQPConnectionError(reason)

    def add_timeout(self, delay_sec, callback):
        deadline = time() + delay_sec
        return self.ioloop.add_timeout(deadline, callback)

    def remove_timeout(self, timeout_id):
        self.ioloop.remove_timeout(timeout_id)

    def _erase_credentials(self):
        pass

    def _flush_outbound(self):
        """
        Call the state manager who will figure out that we need to write.
        """
        self._manage_event_state()

{'fraction_non_alphanumeric': 0.07131236442516269, 'fraction_numerical': 0.0041576283441793205, 'mean_word_length': 2.2231284590736964, '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': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7226970', 'n_tokens_mistral': 3008, 'n_tokens_neox': 2837, 'n_words': 869}