def __init__(self, topic, exchange,
                 type_handlers=None, on_wait=None, url=None,
                 transport=None, transport_options=None,
                 retry_options=None):
        self._topic = topic
        self._exchange_name = exchange
        self._on_wait = on_wait
        self._running = threading.Event()
        self._dispatcher = dispatcher.TypeDispatcher(
            # NOTE(skudriashev): Process all incoming messages only if proxy is
            # running, otherwise requeue them.
            requeue_filters=[lambda data, message: not self.is_running],
            type_handlers=type_handlers)

        ensure_options = self.DEFAULT_RETRY_OPTIONS.copy()
        if retry_options is not None:
            # Override the defaults with any user provided values...
            for k in set(six.iterkeys(ensure_options)):
                if k in retry_options:
                    # Ensure that the right type is passed in...
                    val = retry_options[k]
                    if k in self._RETRY_INT_OPTS:
                        tmp_val = int(val)
                    else:
                        tmp_val = float(val)
                    if tmp_val < 0:
                        raise ValueError("Expected value greater or equal to"
                                         " zero for 'retry_options' %s; got"
                                         " %s instead" % (k, val))
                    ensure_options[k] = tmp_val
        self._ensure_options = ensure_options

        self._drain_events_timeout = DRAIN_EVENTS_PERIOD
        if transport == 'memory' and transport_options:
            polling_interval = transport_options.get('polling_interval')
            if polling_interval is not None:
                self._drain_events_timeout = polling_interval

        # create connection
        self._conn = kombu.Connection(url, transport=transport,
                                      transport_options=transport_options)

        # create exchange
        self._exchange = kombu.Exchange(name=self._exchange_name,
                                        durable=False, auto_delete=True)
dataset/ETHPy150Open openstack/taskflow/taskflow/engines/worker_based/proxy.py/Proxy.__init__/original