jobs.py 文件源码

def _iterate_plugins(self, plugins):
        """Iterates plugins."""
        plugins = iter(plugins)

        def log_plugin_failure(failure, plugin_instance):
            if failure.check(TimeoutError, defer.TimeoutError):
                self._logger.debug("Plugin %s reported a timeout",
                                   plugin_instance.alias, exc_info=True)
                raise AbortedJobError(
                    "Plugin %s reported a timeout" % plugin_instance.alias)
            elif failure.check(SuggestedReschedule):
                self._logger.debug("Plugin %s suggested a reschedule in "
                                   "%d seconds",
                                   plugin_instance, failure.value.delay)
            elif failure.check(db.ResetDBConnectionError):
                pass
            else:
                log_unhandled_failure(self._logger,
                                      failure,
                                      "Plugin %s reported an unhandled failure",
                                      plugin_instance)
            return failure

        def next_plugin(result=None):
            self._raise_if_cancelled()
            try:
                plugin_instance = next(plugins)
            except StopIteration:
                return result

            self._logger.debug("Now calling plugin: %s", plugin_instance)
            self._start_plugin_timer(plugin_instance)

            df = defer.maybeDeferred(plugin_instance.handle)
            df.addErrback(self._stop_plugin_timer)
            df.addErrback(log_plugin_failure, plugin_instance)
            df.addCallback(self._stop_plugin_timer)
            df.addCallback(next_plugin)
            return df

        return next_plugin()