# taskflow.engines.worker_based.worker

# Source code for taskflow.engines.worker_based.worker

    #    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 os
    import platform
    import socket
    import sys

    import futurist
    from oslo_utils import reflection

    from taskflow.engines.worker_based import endpoint
    from taskflow.engines.worker_based import server
    from taskflow import logging
    from taskflow import task as t_task
    from taskflow.utils import banner
    from taskflow.utils import misc
    from taskflow.utils import threading_utils as tu

    LOG = logging.getLogger(__name__)

    class Worker:
        """Worker that can be started on a remote host for handling tasks requests.

        :param url: broker url
        :param exchange: broker exchange name
        :param topic: topic name under which worker is stated
        :param tasks: task list that worker is capable of performing, items in
            the list can be one of the following types; 1, a string naming the
            python module name to search for tasks in or the task class name; 2, a
            python module  to search for tasks in; 3, a task class object that
            will be used to create tasks from.
        :param executor: custom executor object that can used for processing
            requests in separate threads (if not provided one will be created)
        :param threads_count: threads count to be passed to the
                              default executor (used only if an executor is not
                              passed in)
        :param transport: transport to be used (e.g. amqp, memory, etc.)
        :param transport_options: transport specific options (see:
                                  http://kombu.readthedocs.org/ for what these
                                  options imply and are expected to be)
        :param retry_options: retry specific options
                              (see: :py:attr:`~.proxy.Proxy.DEFAULT_RETRY_OPTIONS`)
        """

        def __init__(
            self,
            exchange,
            topic,
            tasks,
            executor=None,
            threads_count=None,
            url=None,
            transport=None,
            transport_options=None,
            retry_options=None,
        ):
            self._topic = topic
            self._executor = executor
            self._owns_executor = False
            if self._executor is None:
                self._executor = futurist.ThreadPoolExecutor(
                    max_workers=threads_count
                )
                self._owns_executor = True
            self._endpoints = self._derive_endpoints(tasks)
            self._exchange = exchange
            self._server = server.Server(
                topic,
                exchange,
                self._executor,
                self._endpoints,
                url=url,
                transport=transport,
                transport_options=transport_options,
                retry_options=retry_options,
            )

        @staticmethod
        def _derive_endpoints(tasks):
            """Derive endpoints from list of strings, classes or packages."""
            derived_tasks = misc.find_subclasses(tasks, t_task.Task)
            return [endpoint.Endpoint(task) for task in derived_tasks]

        @misc.cachedproperty
        def banner(self):
            """A banner that can be useful to display before running."""
            connection_details = self._server.connection_details
            transport = connection_details.transport
            if transport.driver_version:
                transport_driver = "{} v{}".format(
                    transport.driver_name, transport.driver_version
                )
            else:
                transport_driver = transport.driver_name
            try:
                hostname = socket.getfqdn()
            except OSError:
                hostname = "???"
            try:
                pid = os.getpid()
            except OSError:
                pid = "???"
            chapters = {
                'Connection details': {
                    'Driver': transport_driver,
                    'Exchange': self._exchange,
                    'Topic': self._topic,
                    'Transport': transport.driver_type,
                    'Uri': connection_details.uri,
                },
                'Powered by': {
                    'Executor': reflection.get_class_name(self._executor),
                    'Thread count': getattr(self._executor, 'max_workers', "???"),
                },
                'Supported endpoints': [str(ep) for ep in self._endpoints],
                'System details': {
                    'Hostname': hostname,
                    'Pid': pid,
                    'Platform': platform.platform(),
                    'Python': sys.version.split("\n", 1)[0].strip(),
                    'Thread id': tu.get_ident(),
                },
            }
            return banner.make_banner('WBE worker', chapters)

        def run(self, display_banner=True, banner_writer=None):
            """Runs the worker."""
            if display_banner:
                if banner_writer is None:
                    for line in self.banner.splitlines():
                        LOG.info(line)
                else:
                    banner_writer(self.banner)
            self._server.start()

        def wait(self):
            """Wait until worker is started."""
            self._server.wait()

        def stop(self):
            """Stop worker."""
            self._server.stop()
            if self._owns_executor:
                self._executor.shutdown()

    if __name__ == '__main__':
        import argparse
        import logging as log

        parser = argparse.ArgumentParser()
        parser.add_argument("--exchange", required=True)
        parser.add_argument("--connection-url", required=True)
        parser.add_argument("--topic", required=True)
        parser.add_argument("--task", action='append', metavar="TASK", default=[])
        parser.add_argument("-v", "--verbose", action='store_true')
        args = parser.parse_args()
        if args.verbose:
            log.basicConfig(level=logging.DEBUG, format="")
        w = Worker(args.exchange, args.topic, args.task, url=args.connection_url)
        w.run()

this page last updated: 2026-07-16 13:14:25

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