Use details in addition to name to find markers

Signed-off-by: Yves Bastide <yves@botify.com>

Author: ybastide

examples/marker.py

@@ -1,4 +1,5 @@
 from simpleflow import Workflow, futures
+from simpleflow.canvas import Chain
 
 
 class MarkerWorkflow(Workflow):
@@ -11,5 +12,29 @@ def run(self):
         m = self.submit(self.record_marker('marker 1', 'some details'))
         self.submit(self.record_marker('marker 2', "2nd marker's details"))
         futures.wait(m)
-        print('Markers: {}'.format(self.list_markers()))
-        print('Markers, all: {}'.format(self.list_markers(all=True)))
+        markers = self.list_markers()
+        assert 2 == len(markers)
+        print('Markers: {}'.format(markers))
+        markers = self.list_markers(all=True)
+        assert 3 == len(markers)
+        print('Markers, all: {}'.format(markers))
+
+
+class MarkerInChainWorkflow(Workflow):
+    name = 'example'
+    version = 'example'
+    task_list = 'example'
+
+    def run(self):
+        chain = Chain(
+            self.record_marker('marker 1'),
+            self.record_marker('marker 1', 'some details'),
+            self.record_marker('marker 2', "2nd marker's details"),
+        )
+        futures.wait(self.submit(chain))
+        markers = self.list_markers()
+        assert 2 == len(markers)
+        print('Markers: {}'.format(markers))
+        markers = self.list_markers(all=True)
+        assert 3 == len(markers)
+        print('Markers, all: {}'.format(markers))

simpleflow/swf/executor.py

@@ -475,13 +475,20 @@ def find_marker_event(self, a_task, history):
         Get the event corresponding to a activity task, if any.
 
         :param a_task:
-        :type a_task: Marker
+        :type a_task: MarkerTask
         :param history:
         :type history: simpleflow.history.History
         :return:
         :rtype: Optional[dict[str, Any]]
         """
+        json_details = a_task.get_json_details()
         marker_list = history.markers.get(a_task.name)
+        if not marker_list:
+            return None
+        marker_list = filter(
+            lambda m: m['state'] == 'recorded' and m['details'] == json_details,
+            marker_list
+        )
         return marker_list[-1] if marker_list else None
 
     TASK_TYPE_TO_EVENT_FINDER = {
@@ -725,7 +732,7 @@ def _compute_priority(self, priority_set_on_submit, a_task):
         if priority_set_on_submit is not PRIORITY_NOT_SET:
             return priority_set_on_submit
         elif (isinstance(a_task, ActivityTask) and
-                      a_task.activity.task_priority is not PRIORITY_NOT_SET):
+              a_task.activity.task_priority is not PRIORITY_NOT_SET):
             return a_task.activity.task_priority
         elif self._workflow.task_priority is not PRIORITY_NOT_SET:
             return self._workflow.task_priority

simpleflow/swf/task.py

@@ -242,6 +242,9 @@ def schedule(self, *args, **kwargs):
         decision = swf.models.decision.MarkerDecision()
         decision.record(
             self.name,
-            json_dumps(self.details) if self.details is not None else None,
+            self.get_json_details(),
         )
         return [decision]
+
+    def get_json_details(self):
+        return json_dumps(self.details) if self.details is not None else None