Add retry to Flow patterns

Implement dependency checks when adding items to flows:
- retry can't be depended on tasks from it's subflow.
- retry can't provide same values as tasks or other retries

Change-Id: Iffa8e673fd2de39407ae22cd38ad523d484cbba7

1 files changed, 10 insertions, 4 deletions

diff --git a/taskflow/patterns/unordered_flow.py b/taskflow/patterns/unordered_flow.py
index 672a533..efa226e 100644
--- a/taskflow/patterns/unordered_flow.py
+++ b/taskflow/patterns/unordered_flow.py
@@ -29,8 +29,8 @@ class Flow(flow.Flow):
     task/flow outputs (provided names/values).
     """
 
-    def __init__(self, name):
-        super(Flow, self).__init__(name)
+    def __init__(self, name, retry=None):
+        super(Flow, self).__init__(name, retry)
         # NOTE(imelnikov): A unordered flow is unordered, so we use
         # set instead of list to save children, children so that
         # people using it don't depend on the ordering
@@ -43,7 +43,10 @@ class Flow(flow.Flow):
 
         # NOTE(harlowja): check that items to be added are actually
         # independent.
-        provides = self.provides
+        provides = set()
+        for subflow in self:
+            provides.update(subflow.provides)
+
         old_requires = self.requires
         for item in items:
             item_provides = item.provides
@@ -54,7 +57,7 @@ class Flow(flow.Flow):
                     "by other item(s) of unordered flow %(flow)s"
                     % dict(item=item.name, flow=self.name,
                            oo=sorted(bad_provs)))
-            same_provides = provides & item.provides
+            same_provides = (provides | self._retry_provides) & item.provides
             if same_provides:
                 raise exceptions.DependencyFailure(
                     "%(item)s provides %(value)s but is already being"
@@ -79,6 +82,7 @@ class Flow(flow.Flow):
     @property
     def provides(self):
         provides = set()
+        provides.update(self._retry_provides)
         for subflow in self:
             provides.update(subflow.provides)
         return provides
@@ -88,6 +92,8 @@ class Flow(flow.Flow):
         requires = set()
         for subflow in self:
             requires.update(subflow.requires)
+        requires.update(self._retry_requires)
+        requires -= self._retry_provides
         return requires
 
     def __len__(self):