Move Swf specific future semantic in swf.futures

Author: darkjh

simpleflow/executor.py

@@ -15,7 +15,7 @@ def get_actual_value(value):
 
     """
     if isinstance(value, futures.Future):
-        return futures.get_result_or_raise(value)
+        return value.result()
     return value
 
 

simpleflow/swf/executor.py

@@ -10,14 +10,14 @@
 
 from simpleflow import (
     executor,
-    futures,
     exceptions,
     constants,
 )
 from simpleflow.activity import Activity
 from simpleflow.workflow import Workflow
 from simpleflow.history import History
 from simpleflow.swf.task import ActivityTask, WorkflowTask
+from . import futures
 
 logger = logging.getLogger(__name__)
 
@@ -181,15 +181,15 @@ def resume_activity(self, task, event):
         if not future:  # Task in history does not count.
             return None
 
-        if not future.finished:  # Still pending or running...
+        if not future.finished():  # Still pending or running...
             return future
 
-        if future.exception is None:  # Result available!
+        if future.exception() is None:  # Result available!
             return future
 
         if event.get('retry', 0) == task.activity.retry:  # No more retry!
             if task.activity.raises_on_failure:
-                raise exceptions.TaskException(task, future.exception)
+                raise exceptions.TaskException(task, future.exception())
             return future  # with future.exception set.
 
         # Otherwise retry the task by scheduling it again.

simpleflow/swf/futures.py

@@ -0,0 +1,36 @@
+from simpleflow import exceptions
+from simpleflow.futures import (
+    FINISHED,
+    CANCELLED,
+    PENDING,
+    RUNNING,
+    Future as BaseFuture
+)
+
+
+class Future(BaseFuture):
+    """Future with special Simple Workflow semantics
+    """
+    @classmethod
+    def wait(cls):
+        raise exceptions.ExecutionBlocked
+
+    def result(self, timeout=None):
+        with self._condition:
+            if self._state != FINISHED:
+                return self.wait()
+            # TODO what happen if cancelled ???
+            return self._result
+
+    def exception(self, timeout=None):
+        with self._condition:
+            if self._state != FINISHED:
+                return self.wait()
+            return self._exception
+
+    def cancel(self):
+        with self._condition:
+            if self._state == FINISHED:
+                return False
+            self._state = CANCELLED
+            return True

tests/test_dataflow.py

@@ -75,7 +75,7 @@ class TestDefinitionWithInput(TestWorkflow):
     """
     def run(self, a):
         b = self.submit(increment, a)
-        return b.result
+        return b.result()
 
 
 def test_workflow_with_input():
@@ -122,7 +122,7 @@ def run(self):
 
         b = self.submit(double, a)
 
-        return b.result
+        return b.result()
 
 
 def test_workflow_with_two_tasks():
@@ -237,7 +237,7 @@ def run(self, *args, **kwargs):
         a = self.submit(increment, 1)
         b = self.submit(increment, a)
 
-        return b.result
+        return b.result()
 
 
 def test_workflow_with_same_task_called_two_times():
@@ -303,7 +303,7 @@ def run(self, *args, **kwargs):
         a = self.submit(increment, 1)
         a = self.submit(double, a)
 
-        return a.result
+        return a.result()
 
 
 def test_workflow_reuse_same_future():
@@ -364,7 +364,7 @@ def run(self, *args, **kwargs):
         b = self.submit(double, a)
         c = self.submit(increment, a)
 
-        return (b.result, c.result)
+        return b.result(), c.result()
 
 
 def test_workflow_with_two_tasks_same_future():
@@ -490,7 +490,7 @@ class TestDefinitionRetryActivity(TestWorkflow):
     def run(self, *args, **kwargs):
         a = self.submit(increment_retry, 7)
 
-        return a.result
+        return a.result()
 
 
 def test_workflow_retry_activity():
@@ -593,7 +593,7 @@ class TestDefinitionChildWorkflow(TestWorkflow):
     """
     def run(self, x):
         y = self.submit(TestDefinition, x)
-        return y.result
+        return y.result()
 
 
 def test_workflow_with_child_workflow():
@@ -725,7 +725,7 @@ def run(self):
         if result.exception:
             self.fail('error')
 
-        return result.result
+        return result.result()
 
 
 def test_workflow_failed_from_definition():
@@ -771,7 +771,7 @@ class TestDefinitionActivityRaisesOnFailure(OnFailureMixin, TestWorkflow):
 
     """
     def run(self):
-        return self.submit(raise_on_failure).result
+        return self.submit(raise_on_failure).result()
 
 
 def test_workflow_activity_raises_on_failure():
@@ -807,7 +807,7 @@ def test_workflow_activity_raises_on_failure():
 
 class TestOnFailureDefinition(OnFailureMixin, TestWorkflow):
     def run(self):
-        if self.submit(raise_error).exception:
+        if self.submit(raise_error).exception():
             self.fail('FAIL')
 
 
@@ -846,7 +846,7 @@ def run(self):
         b = self.submit(increment, 2)
         c = self.submit(double, b)
 
-        return [a.result, b.result, c.result]
+        return [a.result(), b.result(), c.result()]
 
 
 def test_multiple_scheduled_activities():