def singleton(clazz):
    def constructor(*args, **kws):
    def __new__(cls, name, bases, attrs):
        def getter(self, method_name):
            def method(*args, **kws):
    def register_delegates(delegate_methods, attrs):
            def make_method(method_name):
                def method(self, *args, **kws):
def args2fields(private=True):
    def _field_name(arg_name):
    def _args2fields(constructor):
        def _wrapped_constuctor(self, *args, **kws):
def get_default_args(o):
def parse_context_var(context, variable_name):
    def __getattr__(self, name):
    def __setattr__(self, name, value):
    def __getitem__(self, name):
    def __setitem__(self, name, value):
    def __init__(self, obj):
    def __getattr__(self, attrname):
    def __setattr__(self, attrname, value):
    def __call__(self, *args, **kwds):
class DelegateMeta(type):
    class UnknownAttrError(GirlFriendSysException):
class ObjDictModel(object):
class SafeOperation(object): class CountDownLatch(object):

    """基于计数的闭锁实现
    """

    def __init__(self, count):
        if count <= 0:
            raise InvalidArgumentException(u"count参数必须为正整数")
        self._count = count
        self._condition = threading.Condition()

    def count_down(self):
        with self._condition:
            self._count -= 1
            if self._count <= 0:
                self._condition.notifyAll()

    def await(self):
        with self._condition:
            while self._count > 0:
                self._condition.wait() class InvalidArgumentException(GirlFriendSysException):

    """当函数输入的参数不正确时抛出此异常
    """
    pass class AbstractJob(WorkflowUnit):

    """任务执行单元抽象
    """

    __metaclass__ = ABCMeta

    @property
    def unittype(self):
        return "job"

    @abstractproperty
    def goto(self):
        pass class AbstractFork(WorkflowUnit):

    """Fork单元抽象
    """

    __metaclass__ = ABCMeta

    @property
    def unittype(self):
        return "fork"

    @abstractproperty
    def start_point(self):
        pass

    @abstractproperty
    def end_point(self):
        pass

    @abstractmethod
    def execute(self, units, parrent_context, parrent_listeners=None):
        pass class AbstractJoin(WorkflowUnit):

    """Join单元抽象
    """

    __metaclass__ = ABCMeta

    @property
    def unittype(self):
        return "join"

    @abstractmethod
    def execute(self, context):
        pass class End(WorkflowUnit):

    """结束单元
    """

    STATUS_OK = 0  # 一切顺利,任务成功结束

    STATUS_BAD_REQUEST = 1  # 请求参数错误,任务无法执行

    STATUS_ERROR_HAPPENED = 2  # 服务出错

    def __init__(self, name, status, result=None, execute=None):
        self._name = name
        self._status = status
        self._result = result
        self._execute = execute

    @property
    def name(self):
        return self._name

    @property
    def unittype(self):
        return "end"

    @property
    def status(self):
        """状态码,描述本次工作流的执行结果
        """
        return self._status

    @property
    def result(self):
        """执行结果,工作流执行的最终汇总结果
        """
        return self._result

    def execute(self, context):
        if self._execute:
            self._result = self._execute(context)

    def __str__(self):
        return "<End name={name}, status={status}, result={result}>".format(
            name=self.name,
            status=self.status,
            result=self.result
        ) class ErrorEnd(End):

    """当出现程序异常时返回的结果
    """

    def __init__(self, exc_type, exc_value, tb,
                 name="error_end", result=None, execute=None):
        """
            :param exc_type 异常类型
            :param exc_value 异常值
            :param tb 异常发生的traceback
        """
        End.__init__(
            self,
            name=name,
            status=End.STATUS_ERROR_HAPPENED,
            result=result,
            execute=execute
        )
        self._exc_type = exc_type
        self._exc_value = exc_value
        self._tb = tb

    @property
    def exc_type(self):
        return self._exc_type

    @property
    def exc_value(self):
        return self._exc_value

    @property
    def tb(self):
        return self._tb

    def print_exc(self):
        """打印异常
        """
        traceback.print_exception(self._exc_type, self._exc_value, self._tb) class Job(AbstractJob):

    """最基本的任务单元实现
    """

    def __init__(self, name, plugin=None, caller=None, args=None, goto=None):
        """
          :param name   任务名称，在整个工作流中唯一供跳转声明使用
          :param plugin 使用的插件名称，如果不指定，在没有caller的情况下，任务名称将作为插件名称
          :param caller 可执行逻辑，在不使用插件的时候可以指定一个函数来作为任务逻辑
                        如果同时指定了caller和plugin，那么会抛出UnknowWitchToExecute异常
          :param args   执行插件所需要的参数，可以不指定，在运行时再具体指定
          :param goto   执行完毕后的下一步任务名，可以不指定，自动取任务链上下一个位置
        """

        self._name = name

        # 两者都有指定，抛出异常
        if plugin and caller:
            raise UnknownWitchToExecuteException(
                u"Job '{}' 不能同时指定caller和plugin".format(self._name)
            )

        # 两者都未指定，那么插件名称默认为Job的名称
        if not plugin and not caller:
            self._plugin_name = name
        else:
            self._plugin_name = plugin

        self._caller = caller
        self._args = args
        self._goto = goto

    @property
    def name(self):
        return self._name

    @property
    def plugin_name(self):
        return self._plugin_name

    @property
    def goto(self):
        return self._goto

    @goto.setter
    def goto(self, goto):
        self._goto = goto

    def execute(self, context):
        # 将参数展开
        self._expand_args(context)

        # 如果是生成器，那么迭代执行任务
        if isinstance(self._args, types.GeneratorType):
            result = [self._execute(context, template_args)
                      for template_args in self._args]
        else:
            result = self._execute(context, self._args)

        # 自动将最终计算结果写入Context
        context["{}.result".format(self.name)] = result
        return result

    def _execute(self, context, template_args):
        args = self._get_runtime_args(context, template_args)

        # 获取可执行对象
        executable = self._get_executable(context)

        result = None
        if args is None:
            result = executable(context)
        elif isinstance(args, SequenceCollectionType):
            result = executable(context, *args)
        elif isinstance(args, types.DictType):
            result = executable(context, **args)

        return result

    def _get_executable(self, context):
        if self._caller:
            return self._caller
        plugin = context.plugin(self._plugin_name)
        return plugin.execute

    def _expand_args(self, context):
        # 将函数类型参数展开
        if isinstance(self._args, types.FunctionType):
            self._args = self._args(context)

        # 如果是字符串类型，那么以上下文中的属性作为参数列表
        if isinstance(self._args, types.StringTypes):
            if self._args.startswith("$"):
                self._args = context[self._args[1:]]
            else:
                self._args = context[self._args]

    def _get_runtime_args(self, context, template_args):
        """获取运行时参数
           Job的参数分为两部分，一部分是在workflow声明时指定的参数
           一部分是在运行时通过context指定的参数

           参数支持两种形式，一种是参数列表，一种是字典关键字的形式

           如果是列表，那么将会用context中的列表取代self._args中的列表
           如果是字典，那么将会执行update操作，context中的项将覆盖_args中的项
           如果self._args为None，那么直接使用context中的参数
           如果self._args与context类型不一致，那么抛出异常

           最终将args中的字符串变量名用Context真实值进行替换
        """

        context_args = context.args(self.name)
        if isinstance(context_args, types.FunctionType):
            context_args = context_args(context)

        if isinstance(context_args, types.StringTypes):
            if context_args.startswith("$"):
                context_args = context[context_args[1:]]
            else:
                context_args = context[context_args]

        args_schema = None
        if not context_args:
            args_schema = template_args
        elif not template_args:
            args_schema = context_args
        elif isinstance(template_args, SequenceCollectionType):
            if not isinstance(context_args, SequenceCollectionType):
                raise InvalidArgumentException(
                    u"Job '{}'的初始参数类型跟上下文指定的参数类型不一致".format(self.name)
                )
            args_schema = template_args
            if context_args:
                args_schema = context_args
        elif isinstance(template_args, types.DictType):
            if not isinstance(context_args, types.DictType):
                raise InvalidArgumentException(
                    u"Job '{}'的初始参数类型跟上下文指定的参数类型不一致".format(self.name)
                )
            args_schema = template_args
            if context_args:
                args_schema = dict(template_args)
                args_schema.update(context_args)
        else:
            raise InvalidArgumentException(u"只能接受列表、元组、字典类型的参数对象")

        # 替换为真正的变量
        if not args_schema:
            return None
        elif isinstance(args_schema, SequenceCollectionType):
            return [parse_context_var(context, arg) for arg in args_schema]
        elif isinstance(args_schema, types.DictType):
            return {arg_name: parse_context_var(context, args_schema[arg_name])
                    for arg_name in args_schema} class Context(AbstractContext):

    """基于本地内存的上下文实现
    """

    __metaclass__ = DelegateMeta

    # 以下方法将代理到字典类
    delegate_methods = [
        "__contains__",
        "__delitem__",
        "__len__",
        "__setitem__",
        "get",
        "__iter__",
    ]

    def __init__(self, parrent=None, config=None, args=None,
                 plugin_mgr=None, logger=None, thread_id=None, data=None):
        super(Context, self).__init__()

        if data is None:
            self.delegate = {}
        else:
            self.delegate = data

        self._parrent = parrent
        if parrent is not None:
            self.delegate.update(parrent.delegate)

        self._config = self._extends_parrent(config, parrent, "config")

        if args is None:
            self._args = self._extends_parrent(
                args, parrent, "_args") or {}
        else:
            self._args = args

        self._plugin_mgr = self._extends_parrent(
            plugin_mgr, parrent, "_plugin_mgr", None)

        self._logger = self._extends_parrent(logger, parrent, "logger")

        self._current_unit = None
        self._thread_id = thread_id

    def _extends_parrent(self, arg_value, parrent, field_name,
                         default_value=None):
        """从父上下文继承值
        """
        if arg_value is None and parrent is not None:
            return getattr(parrent, field_name, default_value)
        return arg_value

    def __getitem__(self, property):
        return self.delegate.get(property, None)

    @property
    def config(self):
        """配置对象
        """
        return self._config

    @property
    def logger(self):
        """获取Logger对象
        """
        return self._logger

    @property
    def current_unit(self):
        """当前正在进行的工作单元名称
        """
        return self._current_unit

    @current_unit.setter
    def current_unit(self, current_unit):
        self._current_unit = current_unit

    @property
    def current_unittype(self):
        """当前正在进行的工作单元类型
        """
        return self._current_unittype

    @current_unittype.setter
    def current_unittype(self, current_unittype):
        self._current_unittype = current_unittype

    @property
    def parrent(self):
        return self._parrent

    @property
    def thread_id(self):
        """当前线程标识
        """
        return self._thread_id

    def args(self, job_name):
        """获取某个job运行所需要的参数
        """
        return self._args.get(job_name, None)

    @property
    def plugin_mgr(self):
        return self._plugin_mgr

    def plugin(self, plugin_name):
        return self._plugin_mgr.plugin(plugin_name)

    def __str__(self):
        return "<Context args={args}, config={config}, ctx={ctx}>".format(
            args=self._args,
            config=self._config,
            ctx=self.delegate
        ) class Workflow(AbstractWorkflow):

    """无状态的本地工作流执行引擎实现
    """

    def __init__(self, workflow_list, config=None,
                 plugin_mgr=plugin_manager, context_factory=Context,
                 logger=None, parrent_context=None, thread_id=None):
        """
        :param workflow_list 工作单元列表
        :param config 配置数据
        :param plugin_mgr 插件管理器，默认是plugin自带的entry_points管理器
        :param context_factory 上下文工厂，必须具有config, args, plugin_mgr, parent
                               这四个约定的参数
        :param logger 日志对象
        """

        self._workflow_list = workflow_list
        self._config = config or Config()
        self._plugin_manager = plugin_mgr

        self._units = {}  # 以名称为Key的工作流单元引用字典
        for idx, unit in enumerate(self._workflow_list):
            if unit.name in self._units:
                raise WorkflowUnitExistedException(unit.name)
            self._units[unit.name] = unit
            if unit.unittype == "job" or unit.unittype == "join":
                # 如果未指定goto，那么goto的默认值是下一个节点
                if unit.goto is None:
                    if idx < len(self._workflow_list) - 1:
                        unit.goto = self._workflow_list[idx + 1].name
                    else:
                        unit.goto = "end"
            elif unit.unittype == "fork":
                # 自动设置起始节点
                if unit.start_point is None:
                    if idx < len(self._workflow_list) - 1:
                        unit.start_point = self._workflow_list[idx + 1].name
                    else:
                        raise InvalidArgumentException(
                            u"Fork单元 '{}' 必须指定一个有效的start_point参数")
                # 设置下一步运行的goto节点，如果未指定，则设置最近的join
                if unit.goto is None:
                    for i, next_unit in enumerate(
                            self._workflow_list[idx + 1:], start=idx + 1):
                        if next_unit.unittype == "join":
                            unit.goto = next_unit.name
                            break
                    else:
                        raise InvalidArgumentException(
                            u"Fork单元 '{}' 必须使用一个有效的goto跳转到Join"
                            .format(unit.name))
                # 自动设置结束节点
                if unit.end_point is None:
                    for i, next_unit in enumerate(
                            self._workflow_list[idx + 1:], start=idx + 1):
                        if (
                            next_unit.unittype == "join" and
                            next_unit.name == unit.goto
                        ):
                            # join unit前一个元素
                            unit.end_point = self._workflow_list[i - 1].name
                            break
                    else:
                        raise InvalidArgumentException(
                            u"Fork单元 '{}' 必须指定一个有效的end_point参数".format(
                                unit.name)
                        )

        self._context_factory = context_factory
        self._listeners = []

        # 创建logger
        if logger is None:
            self._logger = create_logger("girlfriend", (stdout_handler(),))
        else:
            self._logger = logger

        self._parrent_context = parrent_context
        self._thread_id = thread_id

    def add_listener(self, listener=None, **kws):
        """为工作流添加监听器，该方法有两种使用方式。

           直接使用: 直接传递一个监听器对象或者一个AbstractListener子类的类型对象
                    当传递监听器对象时，此对象会在每次进行execute时重复使用。
                    当传递类型对象时，每次进行execute都会依照此类型生成一个新的对象
                    类型对象的构造函数参数除self以外必须为空。
           包装函数: 只传递感兴趣的生命周期
                    workflow.add_listener(start=on_start, finish=on_finish)
        """
        if listener:
            if (not isinstance(listener, AbstractListener) and
                    not issubclass(listener, AbstractListener)):
                raise InvalidArgumentException(
                    u"监听器必须是AbstractListener对象或者是其子类类型对象")
            self._listeners.append(listener)
            return
        if kws:
            event_funcs = []
            for event_name in kws:
                if not event_name.startswith("on_"):
                    event_name = "on_" + event_name
                if event_name not in AbstractListener.EVENTS:
                    raise InvalidArgumentException(u"不被支持的事件:" + event_name)
                event_funcs += (event_name, kws[event_name])
            self._listeners.append(AbstractListener.wrap_function(event_funcs))
            return

    def execute(self, args=None, start_point=None, end_point=None, ctrl=None):
        """执行工作流
           :param start_point 工作流起始点，如果为None，
                              那么将从workflow_list第一个元素开始执行
           :param end_point   工作流结束点，如果为None，则按规定节点顺序结束
           :param args 运行时参数
           :param ctrl 会话控制器
        """

        ctrl = SafeOperation(ctrl)

        if start_point:
            start_unit = self._units[start_point]
        else:
            start_unit = self._workflow_list[0]

        if args is None:
            args = {}

        # 构建上下文
        ctx = self._context_factory(
            parrent=self._parrent_context,
            config=self._config,
            args=args,
            plugin_mgr=self._plugin_manager,
            logger=self._logger,
            thread_id=self._thread_id
        )
        listener_objects = {}  # 用来保存每次执行时需要创建新对象的listener

        self._logger.info(u"工作流开始执行，起始点为 '{}'".format(start_unit.name))

        # 执行初始化的listener
        self._execute_listeners("on_start", ctx, listener_objects)

        current_unit = start_unit
        last_result = None
        goto = None
        # 为会话控制器设置当前执行状态为RUNNING
        ctrl.status = AbstractSessionCtrl.STATUS_RUNNING
        while True:
            ctx.current_unit = current_unit.name
            ctx.current_unittype = current_unit.unittype

            # 当前正在执行中的单元
            ctrl.current_unit = current_unit.name

            # 判断停止标记
            if ctrl.stop_mark:
                if ctrl.stop_on is None or ctrl.stop_on == current_unit.name:
                    # 执行所有的on_stop事件
                    ctrl.status = AbstractSessionCtrl.STATUS_STOPPED
                    print "hehehehe"
                    self._execute_listeners("on_stop", ctx, listener_objects)
                    raise WorkflowStoppedException(current_unit.name)

            # 进入新的Unit
            self._logger.info(u"开始执行工作单元 {} [{}]".format(
                current_unit.name, current_unit.unittype))

            self._execute_listeners("on_unit_start", ctx, listener_objects)

            try:
                if current_unit.unittype == "job":
                    last_result = current_unit.execute(ctx)
                    if (
                            end_point is not None and
                            end_point == current_unit.name
                    ):
                        goto = "end"  # 该单元为结束单元
                    else:
                        goto = current_unit.goto
                elif current_unit.unittype == "decision":
                    goto = current_unit.execute(ctx)
                elif current_unit.unittype == "fork":
                    current_unit.execute(
                        self._workflow_list, ctx, self._listeners)
                    goto = current_unit.goto
                elif current_unit.unittype == "join":
                    last_result = current_unit.execute(ctx)
                    goto = current_unit.goto
                elif current_unit.unittype == "end":
                    # 用户指定的结束节点
                    current_unit.execute(ctx)
                    self._execute_listeners("on_unit_finish",
                                            ctx, listener_objects)
                    self._execute_listeners("on_finish", ctx, listener_objects)
                    self._logger.info(u"工作流成功执行完毕")
                    ctrl.mark_finished(result=current_unit.result)
                    return current_unit
            except InvalidArgumentException as e:
                self._logger.exception(u"单元参数错误")
                exc_type, exc_value, tb = sys.exc_info()
                self._execute_on_error_listeners(ctx,
                                                 exc_type, exc_value, tb,
                                                 listener_objects)
                ctrl.mark_exception(exc_info=(exc_type, exc_value, tb))
                return BadRequestEnd(msg=unicode(e))
            except Exception:
                self._logger.exception(u"系统错误，工作流被迫中止")
                exc_type, exc_value, tb = sys.exc_info()
                self._execute_on_error_listeners(ctx,
                                                 exc_type, exc_value, tb,
                                                 listener_objects)
                ctrl.mark_exception(exc_info=(exc_type, exc_value, tb))
                return ErrorEnd(exc_type, exc_value, tb)  # 返回错误的结果
            else:
                # 执行完成事件
                self._execute_listeners("on_unit_finish",
                                        ctx, listener_objects)
                self._logger.info(u"工作单元 {} [{}] 执行完毕".format(
                    current_unit.name, current_unit.unittype))
                if goto == "end":
                    self._execute_listeners("on_finish", ctx, listener_objects)
                    self._logger.info(u"工作流成功执行完毕")
                    ctrl.mark_finished(result=last_result)
                    return OkEnd(result=last_result)  # 将最后一个单元的结果作为默认返回的结果
                else:
                    current_unit = self._units[goto]  # 处理下一个单元

    def _execute_listeners(self, event_name, context, listener_objects):
        """遍历执行监听器
        """
        for idx, listener in enumerate(self._listeners):
            if isinstance(listener, AbstractListener):
                getattr(listener, event_name)(context)
            elif issubclass(listener, AbstractListener):
                listener_obj = self.__create_listener_obj(
                    idx, listener, listener_objects)
                getattr(listener_obj, event_name)(context)

    def _execute_on_error_listeners(self, context,
                                    exc_type, exc_value, tb, listener_objects):
        """遍历执行错误监听器
        """
        for idx, listener in enumerate(self._listeners):
            if isinstance(listener, AbstractListener):
                listener.on_error(context, exc_type, exc_value, tb)
            elif issubclass(listener, AbstractListener):
                listener_obj = self.__create_listener_obj(
                    idx, listener, listener_objects)
                listener_obj.on_error(
                    context, exc_type, exc_value, tb)

    def __create_listener_obj(self, idx, listener_class, listener_objects):
        listener_obj = listener_objects.get(idx)
        if not listener_obj:
            listener_obj = listener_class()
            listener_objects[idx] = listener_obj
        return listener_obj
import sys
import types
import threading
from concurrent.futures import ThreadPoolExecutor
from girlfriend.util.lang import args2fields, SequenceCollectionType
from girlfriend.util.concurrent import CountDownLatch
from girlfriend.exception import InvalidArgumentException
from girlfriend.workflow.protocol import (
    AbstractJob,
    AbstractFork,
    AbstractJoin,
    End,
    ErrorEnd
)
from girlfriend.workflow.gfworkflow import Job, Context, Workflow
                            如果是continue，那么会忽略错误继续执行
        :param error_handler 错误处理器，如果错误处理动作为continue，那么不会调用上层workflow
                             的错误监听器，而是会调用此处的错误处理器
        :param error_default_value 如果是error_action为continue，那么会以该默认值作为错误操作默认结果
        :param goto 要执行的下一个工作单元
        """
        Job.__init__(self, name, plugin, caller, args, goto)
        self._thread_num = thread_num
        self._task_num_per_thread = task_num_per_thread
        self._pool_type, self._sub_join = pool_type, sub_join
        self._result_join, self._error_action = result_join, error_action
        self._error_handler = error_handler
        self._error_default_value = error_default_value

        if self._error_action != "stop" and self._error_action != "continue":
            raise InvalidArgumentException(u"错误处理动作只允许stop或者continue类型")

        try:
            len(args)
        except TypeError:
            if self._task_num_per_thread is None:
                raise InvalidArgumentException(
                    u"args参数为无法预知长度的类型，无法自动分配任务，"
                    u"请使用task_num_per_thread参数来为每个线程分配任务数目"
                )
        else:
            # 根据参数总数目计算每个线程应该分配的任务数
            if self._task_num_per_thread is None:
                args_len = len(self._args)
                if args_len % self._thread_num == 0:
                    self._task_num_per_thread = args_len / self._thread_num
                else:
                    self._task_num_per_thread = args_len / self._thread_num + 1

        self._error_break = False  # 错误中断标记

    def execute(self, context):
        context.logger.info((
            "Concurrent foreach job '{}' begin, "
            "the thread pool size is {}, task/thread is {}"
        ).format(self._name, self._thread_num, self._task_num_per_thread))
        pool = self._pool_type(self._thread_num)
        sub_args = []
        futures = []
        try:
            if isinstance(self._args,
                          (types.FunctionType, types.GeneratorType)):
                self._args = self._args()
            for idx, arg in enumerate(self._args, start=1):
                sub_args.append(arg)
                if idx % self._task_num_per_thread == 0:
                    futures.append(
                        pool.submit(
                            ConcurrentForeachJob._execute,
                            self, context, sub_args)
                    )
                    sub_args = []
            else:
                if sub_args:
                    futures.append(
                        pool.submit(ConcurrentForeachJob._execute,
                                    self, context, sub_args)
                    )
        finally:
            pool.shutdown()

        # 等待futures
        results = [f.result() for f in futures]

        if self._result_join is not None:
            results = self._result_join(context, results)

        context["{}.result".format(self._name)] = results
        return results

    def _execute(self, context, sub_args):
        executable = self._get_executable(context)
        results = []
        for args in sub_args:
            if self._error_break:  # 检查任务是否已被中断
                return
            try:
                result = None
                if args is None:
                    result = executable(context)
                elif isinstance(args, SequenceCollectionType):
                    result = executable(context, *args)
                elif isinstance(args, types.DictType):
                    result = executable(context, **args)
            except Exception as e:
                context.logger.exception(
                    u"并行任务'{}'的子任务运行出错，处理方式为：'{}'".format(
                        self.name, self._error_action))
                if self._error_action == "stop":
                    # rethrow异常，中断工作流
                    self._error_break = True  # 标记任务已中断
                    raise e
                elif self._error_action == "continue":
                    # 调用用户自定义error_handler
                    if self._error_handler is not None:
                        exc_type, exc_value, tb = sys.exc_info()
                        self._error_handler(context, exc_type, exc_value, tb)
                    # 被忽略任务的结果作为None添加到结果列表
                    results.append(self._error_default_value)
            else:
                results.append(result)

        if self._sub_join is not None:
            return self._sub_join(context, results)
        return results


class ConcurrentFork(AbstractFork):

    """该工作单元能够Fork一个新的线程来执行子工作流
    """

    @args2fields()
    def __init__(self, name, thread_num, pool=None,
                 pool_type=ThreadPoolExecutor,

                 start_point=None, end_point=None, context_factory=Context,