# taskflow.engines.worker_based.proxy

# Source code for taskflow.engines.worker_based.proxy

    # -*- coding: utf-8 -*-

    #    Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
    #
    #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    #    not use this file except in compliance with the License. You may obtain
    #    a copy of the License at
    #
    #         http://www.apache.org/licenses/LICENSE-2.0
    #
    #    Unless required by applicable law or agreed to in writing, software
    #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
    #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
    #    License for the specific language governing permissions and limitations
    #    under the License.

    import collections
    import threading

    import kombu
    from kombu import exceptions as kombu_exceptions
    import six

    from taskflow.engines.worker_based import dispatcher
    from taskflow import logging

    LOG = logging.getLogger(__name__)

    # NOTE(skudriashev): A timeout of 1 is often used in environments where
    # the socket can get "stuck", and is a best practice for Kombu consumers.
    DRAIN_EVENTS_PERIOD = 1

    # Helper objects returned when requested to get connection details, used
    # instead of returning the raw results from the kombu connection objects
    # themselves so that a person can not mutate those objects (which would be
    # bad).
    _ConnectionDetails = collections.namedtuple('_ConnectionDetails',
                                                ['uri', 'transport'])
    _TransportDetails = collections.namedtuple('_TransportDetails',
                                               ['options', 'driver_type',
                                                'driver_name', 'driver_version'])

    class Proxy(object):
        """A proxy processes messages from/to the named exchange.

        For **internal** usage only (not for public consumption).
        """

        DEFAULT_RETRY_OPTIONS = {
            # The number of seconds we start sleeping for.
            'interval_start': 1,
            # How many seconds added to the interval for each retry.
            'interval_step': 1,
            # Maximum number of seconds to sleep between each retry.
            'interval_max': 1,
            # Maximum number of times to retry.
            'max_retries': 3,
        }
        """Settings used (by default) to reconnect under transient failures.

        See: http://kombu.readthedocs.org/ (and connection ``ensure_options``) for
        what these values imply/mean...
        """

        # This is the only provided option that should be an int, the others
        # are allowed to be floats; used when we check that the user-provided
        # value is valid...
        _RETRY_INT_OPTS = frozenset(['max_retries'])

        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)

        @property
        def dispatcher(self):
            """Dispatcher internally used to dispatch message(s) that match."""
            return self._dispatcher

        @property
        def connection_details(self):
            """Details about the connection (read-only)."""
            # The kombu drivers seem to use 'N/A' when they don't have a version...
            driver_version = self._conn.transport.driver_version()
            if driver_version and driver_version.lower() == 'n/a':
                driver_version = None
            if self._conn.transport_options:
                transport_options = self._conn.transport_options.copy()
            else:
                transport_options = {}
            transport = _TransportDetails(
                options=transport_options,
                driver_type=self._conn.transport.driver_type,
                driver_name=self._conn.transport.driver_name,
                driver_version=driver_version)
            return _ConnectionDetails(
                uri=self._conn.as_uri(include_password=False),
                transport=transport)

        @property
        def is_running(self):
            """Return whether the proxy is running."""
            return self._running.is_set()

        def _make_queue(self, routing_key, exchange, channel=None):
            """Make a named queue for the given exchange."""
            queue_name = "%s_%s" % (self._exchange_name, routing_key)
            return kombu.Queue(name=queue_name,
                               routing_key=routing_key, durable=False,
                               exchange=exchange, auto_delete=True,
                               channel=channel)

        def publish(self, msg, routing_key, reply_to=None, correlation_id=None):
            """Publish message to the named exchange with given routing key."""
            if isinstance(routing_key, six.string_types):
                routing_keys = [routing_key]
            else:
                routing_keys = routing_key

            # Filter out any empty keys...
            routing_keys = [r_k for r_k in routing_keys if r_k]
            if not routing_keys:
                LOG.warning("No routing key/s specified; unable to send '%s'"
                            " to any target queue on exchange '%s'", msg,
                            self._exchange_name)
                return

            def _publish(producer, routing_key):
                queue = self._make_queue(routing_key, self._exchange)
                producer.publish(body=msg.to_dict(),
                                 routing_key=routing_key,
                                 exchange=self._exchange,
                                 declare=[queue],
                                 type=msg.TYPE,
                                 reply_to=reply_to,
                                 correlation_id=correlation_id)

            def _publish_errback(exc, interval):
                LOG.exception('Publishing error: %s', exc)
                LOG.info('Retry triggering in %s seconds', interval)

            LOG.debug("Sending '%s' message using routing keys %s",
                      msg, routing_keys)
            with kombu.connections[self._conn].acquire(block=True) as conn:
                with conn.Producer() as producer:
                    ensure_kwargs = self._ensure_options.copy()
                    ensure_kwargs['errback'] = _publish_errback
                    safe_publish = conn.ensure(producer, _publish, **ensure_kwargs)
                    for routing_key in routing_keys:
                        safe_publish(producer, routing_key)

        def start(self):
            """Start proxy."""

            def _drain(conn, timeout):
                try:
                    conn.drain_events(timeout=timeout)
                except kombu_exceptions.TimeoutError:
                    pass

            def _drain_errback(exc, interval):
                LOG.exception('Draining error: %s', exc)
                LOG.info('Retry triggering in %s seconds', interval)

            LOG.info("Starting to consume from the '%s' exchange.",
                     self._exchange_name)
            with kombu.connections[self._conn].acquire(block=True) as conn:
                queue = self._make_queue(self._topic, self._exchange, channel=conn)
                callbacks = [self._dispatcher.on_message]
                with conn.Consumer(queues=queue, callbacks=callbacks) as consumer:
                    ensure_kwargs = self._ensure_options.copy()
                    ensure_kwargs['errback'] = _drain_errback
                    safe_drain = conn.ensure(consumer, _drain, **ensure_kwargs)
                    self._running.set()
                    try:
                        while self._running.is_set():
                            safe_drain(conn, self._drain_events_timeout)
                            if self._on_wait is not None:
                                self._on_wait()
                    finally:
                        self._running.clear()

        def wait(self):
            """Wait until proxy is started."""
            self._running.wait()

        def stop(self):
            """Stop proxy."""
            self._running.clear()

this page last updated: 2020-04-10 14:06:26

Except where otherwise noted, this document is licensed under Creative Commons Attribution 3.0 License. See all OpenStack Legal Documents.

found an error? report a bug questions?