# Path: taskflow/engines/worker_based/dispatcher.py
# LOG = logging.getLogger(__name__)
# class Handler(object):
# class TypeDispatcher(object):
#     def __init__(self, process_message, validator=None):
#     def process_message(self):
#     def validator(self):
#     def __init__(self, type_handlers=None, requeue_filters=None):
#     def type_handlers(self):
#     def requeue_filters(self):
#     def _collect_requeue_votes(self, data, message):
#     def _requeue_log_error(self, message, errors):
#     def _process_message(self, data, message, message_type):
#     def on_message(self, data, message):
# 
# Path: taskflow/logging.py
# _BASE = __name__.split(".", 1)[0]
# BLATHER = 5
# TRACE = BLATHER
# CRITICAL = logging.CRITICAL
# DEBUG = logging.DEBUG
# ERROR = logging.ERROR
# FATAL = logging.FATAL
# INFO = logging.INFO
# NOTSET = logging.NOTSET
# WARN = logging.WARN
# WARNING = logging.WARNING
# class _TraceLoggerAdapter(logging.LoggerAdapter):
#     def trace(self, msg, *args, **kwargs):
#     def warn(self, msg, *args, **kwargs):
# def getLogger(name=_BASE, extra=None):
# 
# Path: taskflow/utils/misc.py
# UNKNOWN_HOSTNAME = "<unknown>"
# NUMERIC_TYPES = six.integer_types + (float,)
# _SCHEME_REGEX = re.compile(r"^([A-Za-z][A-Za-z0-9+.-]*):")
# class StrEnum(str, enum.Enum):
# class StringIO(six.StringIO):
# class BytesIO(six.BytesIO):
# class cachedproperty(object):
#     def __new__(cls, *args, **kwargs):
#     def write_nl(self, value, linesep=os.linesep):
#     def reset(self):
# def get_hostname(unknown_hostname=UNKNOWN_HOSTNAME):
# def match_type(obj, matchers):
# def countdown_iter(start_at, decr=1):
# def extract_driver_and_conf(conf, conf_key):
# def reverse_enumerate(items):
# def merge_uri(uri, conf):
# def find_subclasses(locations, base_cls, exclude_hidden=True):
# def pick_first_not_none(*values):
# def parse_uri(uri):
# def disallow_when_frozen(excp_cls):
#     def decorator(f):
#         def wrapper(self, *args, **kwargs):
# def clamp(value, minimum, maximum, on_clamped=None):
# def fix_newlines(text, replacement=os.linesep):
# def binary_encode(text, encoding='utf-8', errors='strict'):
# def binary_decode(data, encoding='utf-8', errors='strict'):
# def _check_decoded_type(data, root_types=(dict,)):
# def decode_msgpack(raw_data, root_types=(dict,)):
# def decode_json(raw_data, root_types=(dict,)):
#     def __init__(self, fget=None, require_lock=True):
#     def __call__(self, fget):
#     def __set__(self, instance, value):
#     def __delete__(self, instance):
#     def __get__(self, instance, owner):
# def millis_to_datetime(milliseconds):
# def get_version_string(obj):
# def sequence_minus(seq1, seq2):
# def as_int(obj, quiet=False):
# def capture_failure():
# def is_iterable(obj):
# def safe_copy_dict(obj):


import functools
from oslo_utils import reflection
from oslo_utils import timeutils
from taskflow.engines.worker_based import dispatcher
from taskflow.engines.worker_based import protocol as pr
from taskflow.engines.worker_based import proxy
from taskflow import logging
from taskflow.types import failure as ft
from taskflow.types import notifier as nt
from taskflow.utils import kombu_utils as ku
from taskflow.utils import misc
# -*- 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.




LOG = logging.getLogger(__name__)


class Server(object):
    """Server implementation that waits for incoming tasks requests."""

    def __init__(self, topic, exchange, executor, endpoints,
                 url=None, transport=None, transport_options=None,
                 retry_options=None):
        type_handlers = {

# Path: taskflow/engines/worker_based/dispatcher.py
# LOG = logging.getLogger(__name__)
# class Handler(object):
# class TypeDispatcher(object):
#     def __init__(self, process_message, validator=None):
#     def process_message(self):
#     def validator(self):
#     def __init__(self, type_handlers=None, requeue_filters=None):
#     def type_handlers(self):
#     def requeue_filters(self):
#     def _collect_requeue_votes(self, data, message):
#     def _requeue_log_error(self, message, errors):
#     def _process_message(self, data, message, message_type):
#     def on_message(self, data, message):
# 
# Path: taskflow/logging.py
# _BASE = __name__.split(".", 1)[0]
# BLATHER = 5
# TRACE = BLATHER
# CRITICAL = logging.CRITICAL
# DEBUG = logging.DEBUG
# ERROR = logging.ERROR
# FATAL = logging.FATAL
# INFO = logging.INFO
# NOTSET = logging.NOTSET
# WARN = logging.WARN
# WARNING = logging.WARNING
# class _TraceLoggerAdapter(logging.LoggerAdapter):
#     def trace(self, msg, *args, **kwargs):
#     def warn(self, msg, *args, **kwargs):
# def getLogger(name=_BASE, extra=None):
# 
# Path: taskflow/utils/misc.py
# UNKNOWN_HOSTNAME = "<unknown>"
# NUMERIC_TYPES = six.integer_types + (float,)
# _SCHEME_REGEX = re.compile(r"^([A-Za-z][A-Za-z0-9+.-]*):")
# class StrEnum(str, enum.Enum):
# class StringIO(six.StringIO):
# class BytesIO(six.BytesIO):
# class cachedproperty(object):
#     def __new__(cls, *args, **kwargs):
#     def write_nl(self, value, linesep=os.linesep):
#     def reset(self):
# def get_hostname(unknown_hostname=UNKNOWN_HOSTNAME):
# def match_type(obj, matchers):
# def countdown_iter(start_at, decr=1):
# def extract_driver_and_conf(conf, conf_key):
# def reverse_enumerate(items):
# def merge_uri(uri, conf):
# def find_subclasses(locations, base_cls, exclude_hidden=True):
# def pick_first_not_none(*values):
# def parse_uri(uri):
# def disallow_when_frozen(excp_cls):
#     def decorator(f):
#         def wrapper(self, *args, **kwargs):
# def clamp(value, minimum, maximum, on_clamped=None):
# def fix_newlines(text, replacement=os.linesep):
# def binary_encode(text, encoding='utf-8', errors='strict'):
# def binary_decode(data, encoding='utf-8', errors='strict'):
# def _check_decoded_type(data, root_types=(dict,)):
# def decode_msgpack(raw_data, root_types=(dict,)):
# def decode_json(raw_data, root_types=(dict,)):
#     def __init__(self, fget=None, require_lock=True):
#     def __call__(self, fget):
#     def __set__(self, instance, value):
#     def __delete__(self, instance):
#     def __get__(self, instance, owner):
# def millis_to_datetime(milliseconds):
# def get_version_string(obj):
# def sequence_minus(seq1, seq2):
# def as_int(obj, quiet=False):
# def capture_failure():
# def is_iterable(obj):
# def safe_copy_dict(obj):


# Path: taskflow/engines/worker_based/server.py
import functools
from oslo_utils import reflection
from oslo_utils import timeutils
from taskflow.engines.worker_based import dispatcher
from taskflow.engines.worker_based import protocol as pr
from taskflow.engines.worker_based import proxy
from taskflow import logging
from taskflow.types import failure as ft
from taskflow.types import notifier as nt
from taskflow.utils import kombu_utils as ku
from taskflow.utils import misc

# -*- 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.




LOG = logging.getLogger(__name__)


class Server(object):
    """Server implementation that waits for incoming tasks requests."""

    def __init__(self, topic, exchange, executor, endpoints,
                 url=None, transport=None, transport_options=None,
                 retry_options=None):
        type_handlers = {

            pr.NOTIFY: dispatcher.Handler(