Have the storage class provide a 'change_flow_state' method

Instead of needing the engine method to lock, fetch, check
and then update the flow state it is more appropriate if the
storage class does all these actions itself (therefore hiding
this change from the engine code, making the engine code
that much smaller).

Change-Id: I2a289c2bcabe76728fa8eb26265ce168abf81b7c

1 files changed, 9 insertions, 13 deletions

diff --git a/taskflow/engines/action_engine/engine.py b/taskflow/engines/action_engine/engine.py
index cc6b1ac..1b0a484 100644
--- a/taskflow/engines/action_engine/engine.py
+++ b/taskflow/engines/action_engine/engine.py
@@ -95,7 +95,6 @@ class ActionEngine(base.Engine):
         self._compilation = None
         self._compiler = compiler.PatternCompiler(flow)
         self._lock = threading.RLock()
-        self._state_lock = threading.RLock()
         self._storage_ensured = False
         # Retries are not *currently* executed out of the engines process
         # or thread (this could change in the future if we desire it to).
@@ -223,18 +222,15 @@ class ActionEngine(base.Engine):
                         failure.Failure.reraise_if_any(it)
 
     def _change_state(self, state):
-        with self._state_lock:
-            old_state = self.storage.get_flow_state()
-            if not states.check_flow_transition(old_state, state):
-                return
-            self.storage.set_flow_state(state)
-        details = {
-            'engine': self,
-            'flow_name': self.storage.flow_name,
-            'flow_uuid': self.storage.flow_uuid,
-            'old_state': old_state,
-        }
-        self.notifier.notify(state, details)
+        moved, old_state = self.storage.change_flow_state(state)
+        if moved:
+            details = {
+                'engine': self,
+                'flow_name': self.storage.flow_name,
+                'flow_uuid': self.storage.flow_uuid,
+                'old_state': old_state,
+            }
+            self.notifier.notify(state, details)
 
     def _ensure_storage(self):
         """Ensure all contained atoms exist in the storage unit."""