summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorJoshua Harlow <harlowja@yahoo-inc.com>2014-03-28 17:34:02 -0700
committerJoshua Harlow <harlowja@yahoo-inc.com>2014-04-01 12:36:14 -0700
commit7b54ac40de0d5d383721a7a09fa0e3e376fa10ba (patch)
treee1528025810f22387f4df15b6559bd13c1abf9d2
parent4366f0f719267892135ec56b4be7403700894a1e (diff)
downloadtaskflow-7b54ac40de0d5d383721a7a09fa0e3e376fa10ba.tar.gz
Allow atoms to save their own state/result0.2
Instead of having storage adjust the internals of the atom detail types (which requires type checking), remove the type checking and let the types themselves decide where to put their own states and results. Change-Id: I397954dc746e9dacb2b65e352d11d8f7f36cdac4
-rw-r--r--taskflow/persistence/logbook.py26
-rw-r--r--taskflow/storage.py20
2 files changed, 30 insertions, 16 deletions
diff --git a/taskflow/persistence/logbook.py b/taskflow/persistence/logbook.py
index d3b0a7a..2de5848 100644
--- a/taskflow/persistence/logbook.py
+++ b/taskflow/persistence/logbook.py
@@ -49,6 +49,10 @@ def _safe_unmarshal_time(when):
return timeutils.unmarshall_time(when)
+def _was_failure(state, result):
+ return state == states.FAILURE and isinstance(result, misc.Failure)
+
+
def _fix_meta(data):
# Handle the case where older schemas allowed this to be non-dict by
# correcting this case by replacing it with a dictionary when a non-dict
@@ -317,6 +321,10 @@ class AtomDetail(object):
def to_dict(self):
"""Translates the internal state of this object to a dictionary."""
+ @abc.abstractmethod
+ def put(self, state, result):
+ """Puts a result (acquired in the given state) into this detail."""
+
def _to_dict_shared(self):
if self.failure:
failure = self.failure.to_dict()
@@ -367,6 +375,15 @@ class TaskDetail(AtomDetail):
self.state = state
self.intention = states.EXECUTE
+ def put(self, state, result):
+ self.state = state
+ if _was_failure(state, result):
+ self.failure = result
+ self.results = None
+ else:
+ self.results = result
+ self.failure = None
+
@classmethod
def from_dict(cls, data):
"""Translates the given data into an instance of this class."""
@@ -416,6 +433,15 @@ class RetryDetail(AtomDetail):
except IndexError as e:
raise exc.NotFound("Last failures not found", e)
+ def put(self, state, result):
+ # Do not clean retry history (only on reset does this happen).
+ self.state = state
+ if _was_failure(state, result):
+ self.failure = result
+ else:
+ self.results.append((result, {}))
+ self.failure = None
+
@classmethod
def from_dict(cls, data):
"""Translates the given data into an instance of this class."""
diff --git a/taskflow/storage.py b/taskflow/storage.py
index 6d3068c..14fd318 100644
--- a/taskflow/storage.py
+++ b/taskflow/storage.py
@@ -53,7 +53,7 @@ class Storage(object):
self._lock = self._lock_cls()
# NOTE(imelnikov): failure serialization looses information,
- # so we cache failures here, in task name -> misc.Failure mapping.
+ # so we cache failures here, in atom name -> failure mapping.
self._failures = {}
for ad in self._flowdetail:
if ad.failure is not None:
@@ -322,24 +322,12 @@ class Storage(object):
"""Put result for atom with id 'uuid' to storage."""
with self._lock.write_lock():
ad = self._atomdetail_by_name(atom_name)
- ad.state = state
+ ad.put(state, data)
if state == states.FAILURE and isinstance(data, misc.Failure):
- # FIXME(harlowja): this seems like it should be internal logic
- # in the atom detail object and not in here. Fix that soon...
- #
- # Do not clean retry history
- if not isinstance(ad, logbook.RetryDetail):
- ad.results = None
- ad.failure = data
+ # NOTE(imelnikov): failure serialization looses information,
+ # so we cache failures here, in atom name -> failure mapping.
self._failures[ad.name] = data
else:
- # FIXME(harlowja): this seems like it should be internal logic
- # in the atom detail object and not in here. Fix that soon...
- if isinstance(ad, logbook.RetryDetail):
- ad.results.append((data, {}))
- else:
- ad.results = data
- ad.failure = None
self._check_all_results_provided(ad.name, data)
self._with_connection(self._save_atom_detail, ad)