Add Workflow.getVersion().

Author: firdaus

cadence/decision_loop.py

@@ -463,6 +463,9 @@ def record_marker(self, marker_name: str, header: Header, details: bytes):
         decision_id = DecisionId(DecisionTarget.MARKER, next_decision_event_id)
         self.decider.add_decision(decision_id, MarkerDecisionStateMachine(id=decision_id, decision=decision))
 
+    def get_version(self, change_id: str, min_supported: int, max_supported: int) -> int:
+        return self.workflow_clock.get_version(change_id, min_supported, max_supported)
+
 @dataclass
 class ReplayDecider:
     execution_id: str
@@ -715,6 +718,7 @@ def handle_marker_recorded(self, event: HistoryEvent):
     def get_optional_decision_event(self, event_id: int) -> HistoryEvent:
         return self.decision_events.get_optional_decision_event(event_id)
 
+
 # noinspection PyUnusedLocal
 def noop(*args):
     pass
@@ -740,7 +744,7 @@ def on_timer_canceled(self: ReplayDecider, event: HistoryEvent):
     EventType.TimerStarted: ReplayDecider.handle_timer_started,
     EventType.TimerCanceled: on_timer_canceled,
     EventType.CancelTimerFailed: ReplayDecider.handle_cancel_timer_failed,
-    EventType.MarkerRecorded: ReplayDecider.handler_marker_recorded
+    EventType.MarkerRecorded: ReplayDecider.handle_marker_recorded
 }
 
 

cadence/tests/test_workflow_get_version.py

@@ -0,0 +1,81 @@
+import sys
+from time import sleep
+
+from cadence.clock_decision_context import DEFAULT_VERSION
+from cadence.workerfactory import WorkerFactory
+from cadence.workflow import workflow_method, Workflow, WorkflowClient
+
+TASK_LIST = "TestWorkflowGetVersion"
+DOMAIN = "sample"
+
+v1_hits = 0
+v2_hits = 0
+
+version_found_in_v2_step_1 = None
+version_found_in_v2_step_2 = None
+
+v2_done = False
+
+
+class TestWorkflowGetVersion:
+    @workflow_method(task_list=TASK_LIST)
+    async def get_greetings(self) -> list:
+        raise NotImplementedError
+
+
+class TestWorkflowGetVersionImplV1(TestWorkflowGetVersion):
+
+    def __init__(self):
+        pass
+
+    async def get_greetings(self):
+        global v1_hits
+        v1_hits += 1
+        await Workflow.sleep(60)
+
+
+class TestWorkflowGetVersionImplV2(TestWorkflowGetVersion):
+
+    def __init__(self):
+        pass
+
+    async def get_greetings(self):
+        global v2_hits
+        global version_found_in_v2_step_1, version_found_in_v2_step_2
+        global v2_done
+        v2_hits += 1
+
+        version_found_in_v2_step_1 = Workflow.get_version("first-item", DEFAULT_VERSION, 2)
+        await Workflow.sleep(60)
+        version_found_in_v2_step_2 = Workflow.get_version("first-item", DEFAULT_VERSION, 2)
+        v2_done = True
+
+
+def test_workflow_workflow_get_version():
+    global v1_hits, v2_hits
+    factory = WorkerFactory("localhost", 7933, DOMAIN)
+    worker = factory.new_worker(TASK_LIST)
+    worker.register_workflow_implementation_type(TestWorkflowGetVersionImplV1)
+    factory.start()
+
+    client = WorkflowClient.new_client(domain=DOMAIN)
+    workflow: TestWorkflowGetVersion = client.new_workflow_stub(TestWorkflowGetVersion)
+
+    client.start(workflow.get_greetings)
+    while v1_hits == 0:
+        print(".", end="")
+        sleep(2)
+
+    worker.register_workflow_implementation_type(TestWorkflowGetVersionImplV2)
+
+    while not v2_done:
+        print(".", end="")
+        sleep(2)
+
+    assert v1_hits == 1
+    assert v2_hits == 1
+    assert version_found_in_v2_step_1 == DEFAULT_VERSION
+    assert version_found_in_v2_step_2 == 2
+
+    print("Stopping workers")
+    worker.stop(background=True)

cadence/tests/test_workflow_get_version_single.py

@@ -0,0 +1,47 @@
+import sys
+from time import sleep
+
+from cadence.clock_decision_context import DEFAULT_VERSION
+from cadence.workerfactory import WorkerFactory
+from cadence.workflow import workflow_method, Workflow, WorkflowClient
+
+TASK_LIST = "TestWorkflowGetVersionSingle"
+DOMAIN = "sample"
+
+version_found_in_step_1 = None
+version_found_in_step_2 = None
+
+
+class TestWorkflowGetVersionSingle:
+    @workflow_method(task_list=TASK_LIST)
+    async def get_greetings(self) -> list:
+        raise NotImplementedError
+
+
+class TestWorkflowGetVersionSingleImpl(TestWorkflowGetVersionSingle):
+
+    def __init__(self):
+        pass
+
+    async def get_greetings(self):
+        global version_found_in_step_1, version_found_in_step_2
+        version_found_in_step_1 = Workflow.get_version("first-item", DEFAULT_VERSION, 2)
+        await Workflow.sleep(60)
+        version_found_in_step_2 = Workflow.get_version("first-item", DEFAULT_VERSION, 2)
+
+
+def test_workflow_workflow_get_version_single():
+    factory = WorkerFactory("localhost", 7933, DOMAIN)
+    worker = factory.new_worker(TASK_LIST)
+    worker.register_workflow_implementation_type(TestWorkflowGetVersionSingleImpl)
+    factory.start()
+
+    client = WorkflowClient.new_client(domain=DOMAIN)
+    workflow: TestWorkflowGetVersionSingle = client.new_workflow_stub(TestWorkflowGetVersionSingle)
+    workflow.get_greetings()
+
+    assert version_found_in_step_1 == 2
+    assert version_found_in_step_2 == 2
+
+    print("Stopping workers")
+    worker.stop(background=True)

cadence/workflow.py

@@ -71,6 +71,15 @@ def new_random() -> random.Random:
         task: ITask = ITask.current()
         return task.decider.decision_context.new_random()
 
+    @staticmethod
+    def get_version(change_id: str, min_supported: int, max_supported: int):
+        from cadence.decision_loop import ITask
+        from cadence.decision_loop import DecisionContext
+        task: ITask = ITask.current()
+        decision_context: DecisionContext = task.decider.decision_context
+        return decision_context.get_version(change_id, min_supported, max_supported)
+
+
 class WorkflowStub:
     pass