def retry(*dargs, **dkw):
    defaults = dict(
        retry_on_exception=retry_filter,
        wait_exponential_multiplier=500,
        wait_exponential_max=15000,
    )

    if len(dargs) == 1 and callable(dargs[0]):
        return retrying.retry(**defaults)(dargs[0])
    else:
        dkw = dict(defaults, **dkw)
        return retrying.retry(*dargs, **dkw)

def retry_on_conflict(func):
    def is_conflict(ex):
        return isinstance(ex, exception.ConcurrentTransaction)
    wrapper = retrying.retry(stop_max_attempt_number=11,
                             wait_random_min=0.0, wait_random_max=2.0,
                             retry_on_exception=is_conflict)
    return wrapper(func)

def retry_database(fn):
    """Decorator to keep probing the database untill you succeed."""
    _check_exception = functools.partial(check_exception, valid_exc=sa.exc.OperationalError)
    r = retry(
        retry_on_exception=_check_exception, wait_exponential_multiplier=1000,
        wait_exponential_max=60000, stop_max_attempt_number=7)
    return r(fn)

def _safe_mongo_call(max_retries, retry_interval):
    return retrying.retry(
        retry_on_exception=lambda e: isinstance(
            e, pymongo.errors.AutoReconnect),
        wait_fixed=retry_interval * 1000,
        stop_max_attempt_number=max_retries if max_retries >= 0 else None
    )

def rety_subprocess(fn):
    r = retry(
        retry_on_exception=lambda exc:
            _check_exception(exc, valid_exc=MySubprocessError),
        wait_exponential_multiplier=1000,
        wait_exponential_max=60000,
        stop_max_attempt_number=7)
    return r(fn)

def mark(f):
    """Marks an entire test as eventually consistent and retries."""
    __tracebackhide__ = True
    return retry(
        wait_exponential_multiplier=100,
        wait_exponential_max=1500,
        stop_max_attempt_number=5,
        retry_on_exception=_retry_on_exception(AssertionError))(f)

 def wrapped(*args, **kwargs):
     self = args[0]
     new_fn = retrying.retry(
         retry_on_result=_ofport_result_pending,
         stop_max_delay=self.vsctl_timeout * 1000,
         wait_exponential_multiplier=10,
         wait_exponential_max=1000,
         retry_on_exception=lambda _: False)(fn)
     return new_fn(*args, **kwargs)

    def test_retry(self):
        a = [1, 2]
        k = {'a': 3, 'b': 4, 'cc': 5}

        self.mox.StubOutWithMock(retrying, 'retry')
        retrying.retry(
            *a,
            retry_on_exception=mox.IgnoreArg(),
            wait_random_min=mox.IgnoreArg(),
            wait_random_max=mox.IgnoreArg(),
            **k
        ).AndReturn(lambda func: func)
        func = self.mox.CreateMockAnything()
        self.mox.StubOutWithMock(functools, 'wraps')
        functools.wraps(func).AndReturn(lambda func: func)
        func()
        self.mox.ReplayAll()
        orm.retry(*a, **k)(func)()

def retry_ssh(fn):
    """Retry doing something over an ssh connection."""
    _check_exception = functools.partial(check_exception, valid_exc=paramiko.SSHException)
    wrapper = retry(
        retry_on_exception=_check_exception,
        wait_exponential_multiplier=1_000,
        wait_exponential_max=60_000,
        stop_max_attempt_number=7)
    return wrapper(fn)

def mark(f):
    """Marks an entire test as eventually consistent and retries."""
    __tracebackhide__ = True
    return retry(
        wait_exponential_multiplier=WAIT_EXPONENTIAL_MULTIPLIER,
        wait_exponential_max=WAIT_EXPONENTIAL_MAX_DEFAULT,
        stop_max_attempt_number=STOP_MAX_ATTEMPT_NUMBER_DEFAULT,
        retry_on_exception=_retry_on_exception(
            (AssertionError, exceptions.GoogleCloudError)))(f)

def retry_archive(fn):
    """Decorator to keep probing the database untill you succeed."""
    from retrying import retry
    from elaspic import errors
    r = retry(
        retry_on_exception=lambda exc:
            _check_exception(exc, valid_exc=errors.Archive7zipError),
        wait_fixed=2000,
        stop_max_attempt_number=2)
    return r(fn)

    def reply(self, reply=None, failure=None):
        """Send back reply to the RPC client
        :param reply: Dictionary, reply. In case of exception should be None
        :param failure: Tuple, should be a sys.exc_info() tuple.
            Should be None if RPC request was successfully processed.

        :return RpcReplyPikaIncomingMessage, message with reply
        """

        if self.reply_q is None:
            return

        reply_outgoing_message = RpcReplyPikaOutgoingMessage(
            self._pika_engine, self.msg_id, reply=reply, failure_info=failure,
            content_type=self._content_type,
            content_encoding=self._content_encoding
        )

        def on_exception(ex):
            if isinstance(ex, pika_drv_exc.ConnectionException):
                LOG.warning(
                    "Connectivity related problem during reply sending. %s",
                    ex
                )
                return True
            else:
                return False

        retrier = retrying.retry(
            stop_max_attempt_number=(
                None if self._pika_engine.rpc_reply_retry_attempts == -1
                else self._pika_engine.rpc_reply_retry_attempts
            ),
            retry_on_exception=on_exception,
            wait_fixed=self._pika_engine.rpc_reply_retry_delay * 1000,
        ) if self._pika_engine.rpc_reply_retry_attempts else None

        try:
            timeout = (None if self.expiration_time is None else
                       max(self.expiration_time - time.time(), 0))
            with timeutils.StopWatch(duration=timeout) as stopwatch:
                reply_outgoing_message.send(
                    reply_q=self.reply_q,
                    stopwatch=stopwatch,
                    retrier=retrier
                )
            LOG.debug(
                "Message [id:'%s'] replied to '%s'.", self.msg_id, self.reply_q
            )
        except Exception:
            LOG.exception(
                "Message [id:'%s'] wasn't replied to : %s", self.msg_id,
                self.reply_q
            )

 def create_image(self, name):
     """
     Create a new image based on this instance.
     """
     image_id = self._ec2_instance.create_image(name)
     # Sometimes, the image takes a while to register, so retry a few times
     # if the image cannot be found
     retry_decorator = retry(retry_on_result=lambda result: result is None,
                             stop_max_attempt_number=3, wait_fixed=1000)
     image = retry_decorator(self._provider.compute.images.get)(image_id)
     return image

def retry_database(fn):
    """Decorator to keep probing the database untill you succeed."""
    from retrying import retry
    import sqlalchemy as sa
    r = retry(
        retry_on_exception=lambda exc:
            _check_exception(exc, valid_exc=sa.exc.OperationalError),
        wait_exponential_multiplier=1000,
        wait_exponential_max=60000,
        stop_max_attempt_number=7)
    return r(fn)

    def reply(self, reply=None, failure=None, log_failure=True):
        """Send back reply to the RPC client
        :param reply: Dictionary, reply. In case of exception should be None
        :param failure: Tuple, should be a sys.exc_info() tuple.
            Should be None if RPC request was successfully processed.
        :param log_failure: Boolean, not used in this implementation.
            It present here to be compatible with driver API

        :return RpcReplyPikaIncomingMessage, message with reply
        """

        if self.reply_q is None:
            return

        reply_outgoing_message = RpcReplyPikaOutgoingMessage(
            self._pika_engine, self.msg_id, reply=reply, failure_info=failure,
            content_type=self._content_type,
            content_encoding=self._content_encoding
        )

        def on_exception(ex):
            if isinstance(ex, pika_drv_exc.ConnectionException):
                LOG.warn(str(ex))
                return True
            else:
                return False

        retrier = retrying.retry(
            stop_max_attempt_number=(
                None if self._pika_engine.rpc_reply_retry_attempts == -1
                else self._pika_engine.rpc_reply_retry_attempts
            ),
            retry_on_exception=on_exception,
            wait_fixed=self._pika_engine.rpc_reply_retry_delay * 1000,
        ) if self._pika_engine.rpc_reply_retry_attempts else None

        try:
            reply_outgoing_message.send(
                reply_q=self.reply_q,
                expiration_time=self.expiration_time,
                retrier=retrier
            )
            LOG.debug(
                "Message [id:'{}'] replied to '{}'.".format(
                    self.msg_id, self.reply_q
                )
            )
        except Exception:
            LOG.exception(
                "Message [id:'{}'] wasn't replied to : {}".format(
                    self.msg_id, self.reply_q
                )
            )

 def retryable(self, function):
     @wraps(function)
     def wrapper(*args, **kwargs):
         self.maybe_wait()
         response = function(*args, **kwargs)
         backoff_interval = response.data.get('backoff', 0)
         self.set_next_request_minimum_time(backoff_interval)
         return response
     return retry(
         retry_on_exception=BackoffStrategy._retry_on_throttle_error,
         **self._retry_kwargs
     )(wrapper)

    def get_publishers_retrying(self):
        """Retry until at least one publisher appears"""

        def retry_if_empty(publishers):
            return not publishers

        _retry = retrying.retry(retry_on_result=retry_if_empty)

        @_retry
        def _get_publishers():
            return self.get_publishers()

        return _get_publishers()

def with_retry(cls, methods):
    """
    Wraps the given list of methods in a class with an exponential-back
    retry mechanism.
    """
    retry_with_backoff = retry(
        retry_on_exception=lambda e: isinstance(e, BotoServerError),
        wait_exponential_multiplier=1000,
        wait_exponential_max=10000
    )
    for method in methods:
        m = getattr(cls, method, None)
        if isinstance(m, collections.Callable):
            setattr(cls, method, retry_with_backoff(m))
    return cls

    def __init__(self, url, http_method='GET', parser=None, stop_max_attempt_number=RETRY_COUNT, **kwargs):
        self.url = url
        for name, value in kwargs.items():
            if isinstance(value, list) or isinstance(value, tuple):
                kwargs[name] = ','.join(str(i) for i in value)
            elif isinstance(value, datetime):
                kwargs[name] = value.isoformat()
        self.parser = parser
        self.params = kwargs
        self.http_method = http_method
        self._data = None
        self.error = None

        # Retry only if SOURCE_NOT_AVAILABLE error
        self._fetch_data = retry(
            stop_max_attempt_number=stop_max_attempt_number,
            retry_on_exception=lambda ex: isinstance(ex, RequestError) and ex.code == 504
        )(self._fetch_data)

    def send_notification(self, target, ctxt, message, version, retry=None):
        if retry is None:
            retry = self._pika_engine.default_notification_retry_attempts

        def on_exception(ex):
            if isinstance(ex, (pika_drv_exc.ExchangeNotFoundException,
                               pika_drv_exc.RoutingException)):
                LOG.warning("Problem during sending notification. %s", ex)
                try:
                    self._declare_notification_queue_binding(target)
                except pika_drv_exc.ConnectionException as e:
                    LOG.warning("Problem during declaring notification queue "
                                "binding. %s", e)
                return True
            elif isinstance(ex, (pika_drv_exc.ConnectionException,
                                 pika_drv_exc.MessageRejectedException)):
                LOG.warning("Problem during sending notification. %s", ex)
                return True
            else:
                return False

        retrier = retrying.retry(
            stop_max_attempt_number=(None if retry == -1 else retry),
            retry_on_exception=on_exception,
            wait_fixed=self._pika_engine.notification_retry_delay * 1000,
        )

        msg = pika_drv_msg.PikaOutgoingMessage(self._pika_engine, message,
                                               ctxt)
        return msg.send(
            exchange=(
                target.exchange or
                self._pika_engine.default_notification_exchange
            ),
            routing_key=target.topic,
            confirm=True,
            mandatory=True,
            persistent=self._pika_engine.notification_persistence,
            retrier=retrier
        )