Merge "Cleanup engine base class"

taskflow/engines/action_engine/engine.py

@@ -62,8 +62,6 @@ class ActionEngine(base.EngineBase):
         self._state_lock = threading.RLock()
         self._task_executor = None
         self._task_action = None
-        self.notifier = misc.TransitionNotifier()
-        self.task_notifier = misc.TransitionNotifier()
 
     def _revert(self, current_failure=None):
         self._change_state(states.REVERTING)
@@ -85,13 +83,6 @@ class ActionEngine(base.EngineBase):
         return "%s: %s" % (reflection.get_class_name(self), id(self))
 
     def suspend(self):
-        """Attempts to suspend the engine.
-
-        If the engine is currently running tasks then this will attempt to
-        suspend future work from being started (currently active tasks can
-        not currently be preempted) and move the engine into a suspend state
-        which can then later be resumed from.
-        """
         if not self._compiled:
             raise exc.InvariantViolation("Can not suspend an engine"
                                          " which has not been compiled")
@@ -173,10 +164,6 @@ class ActionEngine(base.EngineBase):
 
     @lock_utils.locked
     def compile(self):
-        """Compiles the contained flow into a structure which the engine can
-        use to run or if this can not be done then an exception is thrown
-        indicating why this compilation could not be achieved.
-        """
         if self._compiled:
             return
         task_graph = flow_utils.flatten(self._flow)

taskflow/engines/base.py

@@ -21,6 +21,8 @@ import abc
 
 import six
 
+from taskflow.utils import misc
+
 
 @six.add_metaclass(abc.ABCMeta)
 class EngineBase(object):
@@ -29,16 +31,43 @@ class EngineBase(object):
     def __init__(self, flow, flow_detail, backend, conf):
         self._flow = flow
         self._flow_detail = flow_detail
-        self.storage = self._storage_cls(flow_detail, backend)
+        self._backend = backend
+        if not conf:
+            self._conf = {}
+        else:
+            self._conf = dict(conf)
+        self._storage = None
+        self.notifier = misc.TransitionNotifier()
+        self.task_notifier = misc.TransitionNotifier()
+
+    @property
+    def storage(self):
+        """The storage unit for this flow."""
+        if self._storage is None:
+            self._storage = self._storage_cls(self._flow_detail, self._backend)
+        return self._storage
 
     @abc.abstractproperty
     def _storage_cls(self):
-        """Storage class"""
+        """Storage class that will be used to generate storage objects"""
 
     @abc.abstractmethod
     def compile(self):
-        """Check the flow and convert it to internal representation"""
+        """Compiles the contained flow into a structure which the engine can
+        use to run or if this can not be done then an exception is thrown
+        indicating why this compilation could not be achieved.
+        """
 
     @abc.abstractmethod
     def run(self):
-        """Run the flow"""
+        """Runs the flow in the engine to completion (or die trying)."""
+
+    @abc.abstractmethod
+    def suspend(self):
+        """Attempts to suspend the engine.
+
+        If the engine is currently running tasks then this will attempt to
+        suspend future work from being started (currently active tasks can
+        not currently be preempted) and move the engine into a suspend state
+        which can then later be resumed from.
+        """