<commit_before># TODO:
# - review locking, race conditions, deadlock
# - get values from proper config
# - proper threadpool submissions

import time
from threading import Lock, RLock, Condition
import weakref

from connection import MAX_STREAM_PER_CONNECTION, ConnectionException


class BusyConnectionException(Exception):
    pass


class AuthenticationException(Exception):
    pass


class TimeoutException(Exception):
    pass


class Host(object):

    def __init__(self, inet_address, conviction_policy_factory):
        if inet_address is None:
            raise ValueError("inet_address may not be None")
        if conviction_policy_factory is None:
            raise ValueError("conviction_policy_factory may not be None")

        self._datacenter = None
        self._rack = None

        self.address = inet_address
        self.monitor = HealthMonitor(conviction_policy_factory(self))

        self._reconnection_handler = None
        self._reconnection_lock = Lock()

    @property
    def datacenter(self):
        return self._datacenter

    @property
    def rack(self):
        return self._rack

    def set_location_info(self, datacenter, rack):
        self._datacenter = datacenter
        self._rack = rack

    def get_and_set_reconnection_handler(self, new_handler):
        with self._reconnection_lock:
            old = self._reconnection_handler
            self._reconnection_handler = new_handler
            return old

    def __eq__(self, other):
        if not isinstance(other, Host):
            return False

        return self.address == other.address

    def __str__(self):
        return self.address

    def __repr__(self):
        return "<%s: %s>" % (self.__class__.__name__, self.address)


class _ReconnectionHandler(object):
    """
    Abstract class for attempting reconnections with a given
    schedule and scheduler.
    """

    def __init__(self, scheduler, schedule, callback, *callback_args, **callback_kwargs):
        self.scheduler = scheduler
        self.schedule = schedule
        self.callback = callback
        self.callback_args = callback_args
        self.callback_kwargs = callback_kwargs
        self._cancelled = False

    def start(self):
        if self._cancelled:
            return

        # TODO cancel previous reconnection handlers? That's probably the job
        # of whatever created this.

        first_delay = self.schedule.get_next_delay()
        self.scheduler.schedule(first_delay, self.run)

    def run(self):
        if self._cancelled:
            return

        # TODO wait for readyForNext?
        try:
            self.on_reconnection(self.try_reconnect())
        except Exception, exc:
            next_delay = self.schedule.get_next_delay()
            if self.on_exception(exc, next_delay):
                self.scheduler.schedule(next_delay, self.run)
        else:
            self.callback(*self.callback_args, **self.callback_kwargs)

    def cancel(self):
        self._cancelled = True

    def try_reconnect(self):
        """
        Subclasses must implement this method.  It should attempt to
        open a new Connection and return it; if a failure occurs, an
        Exception should be raised.
        """
        raise NotImplemented()
<commit_msg>Reconnection schedule is a generator, so use next()<commit_after># TODO:
# - review locking, race conditions, deadlock
# - get values from proper config
# - proper threadpool submissions

import time
from threading import Lock, RLock, Condition
import weakref

from connection import MAX_STREAM_PER_CONNECTION, ConnectionException


class BusyConnectionException(Exception):
    pass


class AuthenticationException(Exception):
    pass


class TimeoutException(Exception):
    pass


class Host(object):

    def __init__(self, inet_address, conviction_policy_factory):
        if inet_address is None:
            raise ValueError("inet_address may not be None")
        if conviction_policy_factory is None:
            raise ValueError("conviction_policy_factory may not be None")

        self._datacenter = None
        self._rack = None

        self.address = inet_address
        self.monitor = HealthMonitor(conviction_policy_factory(self))

        self._reconnection_handler = None
        self._reconnection_lock = Lock()

    @property
    def datacenter(self):
        return self._datacenter

    @property
    def rack(self):
        return self._rack

    def set_location_info(self, datacenter, rack):
        self._datacenter = datacenter
        self._rack = rack

    def get_and_set_reconnection_handler(self, new_handler):
        with self._reconnection_lock:
            old = self._reconnection_handler
            self._reconnection_handler = new_handler
            return old

    def __eq__(self, other):
        if not isinstance(other, Host):
            return False

        return self.address == other.address

    def __str__(self):
        return self.address

    def __repr__(self):
        return "<%s: %s>" % (self.__class__.__name__, self.address)


class _ReconnectionHandler(object):
    """
    Abstract class for attempting reconnections with a given
    schedule and scheduler.
    """

    def __init__(self, scheduler, schedule, callback, *callback_args, **callback_kwargs):
        self.scheduler = scheduler
        self.schedule = schedule
        self.callback = callback
        self.callback_args = callback_args
        self.callback_kwargs = callback_kwargs
        self._cancelled = False

    def start(self):
        if self._cancelled:
            return

        # TODO cancel previous reconnection handlers? That's probably the job
        # of whatever created this.

        first_delay = self.schedule.next()
        self.scheduler.schedule(first_delay, self.run)

    def run(self):
        if self._cancelled:
            return

        # TODO wait for readyForNext?
        try:
            self.on_reconnection(self.try_reconnect())
        except Exception, exc:
            next_delay = self.schedule.next()
            if self.on_exception(exc, next_delay):
                self.scheduler.schedule(next_delay, self.run)
        else:
            self.callback(*self.callback_args, **self.callback_kwargs)

    def cancel(self):
        self._cancelled = True

    def try_reconnect(self):
        """
        Subclasses must implement this method.  It should attempt to
        open a new Connection and return it; if a failure occurs, an
        Exception should be raised.
        """
        raise NotImplemented()

{'fraction_non_alphanumeric': 0.07321156773211568, 'fraction_numerical': 0.0, 'mean_word_length': 2.532795698924731, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': '7083684', 'n_tokens_mistral': 1773, 'n_tokens_neox': 1704, 'n_words': 487}