Merge pull request #826 from fjtirado/Fix_#677

[Fix #677] Atomic produce and consume actions
serverlessworkflow · Mar 20, 2024 · 7ae6420 · 7ae6420
2 parents 9f79f8f + 30c8fec
commit 7ae6420
Show file tree

Hide file tree

Showing 7 changed files with 122 additions and 99 deletions.
diff --git a/examples/README_TEMPLATE.md b/examples/README_TEMPLATE.md
@@ -1091,7 +1091,6 @@ events:
   - contextAttributeName: accountId
 - name: ConfirmationCompletedEvent
   type: payment.confirmation
-  kind: produced
 
 ```
 

diff --git a/examples/event-based-service-invocation.json b/examples/event-based-service-invocation.json
@@ -9,14 +9,12 @@
         {
             "name": "make-vet-appointment",
             "source": "VetServiceSource",
-            "type": "events.vet.appointments",
-            "kind": "produced"
+            "type": "events.vet.appointments"
         },
         {
             "name": "vet-appointment-info",
             "source": "VetServiceSource",
-            "type": "events.vet.appointments",
-            "kind": "consumed"
+            "type": "events.vet.appointments"
         }
     ],
     "states": [
@@ -26,10 +24,15 @@
             "actions": [
                 {
                     "name": "make-appointment-action",
-                    "eventRef": {
-                        "produceEventRef": "make-vet-appointment",
-                        "data": "${ .patientInfo }",
-                        "consumeEventRef": "vet-appointment-info"
+                    "publish": {
+                        "event": "make-vet-appointment",
+                        "data": "${ .patientInfo }"
+                    }
+                },
+                {
+                    "name": "wait-appointement-confirmation",
+                    "subscribe": {
+                        "event": "vet-appointment-info"
                     },
                     "actionDataFilter": {
                         "results": "${ .appointmentInfo }"

diff --git a/examples/purchase-order-deadline.json b/examples/purchase-order-deadline.json
@@ -137,13 +137,11 @@
         },
         {
             "name": "order-finished-event",
-            "type": "my.company.orders",
-            "kind": "produced"
+            "type": "my.company.orders"
         },
         {
             "name": "order-cancelled-event",
-            "type": "my.company.orders",
-            "kind": "produced"
+            "type": "my.company.orders"
         }
     ],
     "functions": [

diff --git a/examples/send-cloudevent-on-workflow-completion.json b/examples/send-cloudevent-on-workflow-completion.json
@@ -7,8 +7,7 @@
     "events": [
         {
             "name": "provisioning-complete-event",
-            "type": "provisionCompleteType",
-            "kind": "produced"
+            "type": "provisionCompleteType"
         }
     ],
     "functions": [

diff --git a/schema/events.json b/schema/events.json
@@ -43,15 +43,6 @@
           "type": "string",
           "description": "CloudEvent type"
         },
-        "kind": {
-          "type": "string",
-          "enum": [
-            "consumed",
-            "produced"
-          ],
-          "description": "Defines the CloudEvent as either 'consumed' or 'produced' by the workflow. Default is 'consumed'",
-          "default": "consumed"
-        },
         "correlation": {
           "type": "array",
           "description": "CloudEvent correlation definitions",

diff --git a/schema/workflow.json b/schema/workflow.json
@@ -413,9 +413,13 @@
           "description": "References a function to be invoked",
           "$ref": "#/definitions/functionref"
         },
-        "eventRef": {
-          "description": "References a `produce` and `consume` reusable event definitions",
-          "$ref": "#/definitions/eventref"
+        "publish": {
+          "description": "Publish an event",
+          "$ref": "#/definitions/publish"
+        },
+        "subscribe": {
+          "description": "Subscribe to an event channel",
+          "$ref": "#/definitions/subscribe"
         },
         "subFlowRef": {
           "description": "References a sub-workflow to invoke",
@@ -468,7 +472,13 @@
         {
           "required": [
             "name",
-            "eventRef"
+            "publish"
+          ]
+        },
+        {
+          "required": [
+            "name",
+            "subscribe"
           ]
         },
         {
@@ -521,30 +531,21 @@
         }
       ]
     },
-    "eventref": {
+    "publish": {
       "type": "object",
-      "description": "Event References",
+      "description": "Publish an event",
       "properties": {
-        "produceEventRef": {
+        "event": {
           "type": "string",
-          "description": "Reference to the unique name of a 'produced' event definition",
+          "description": "Reference to the unique name of a 'published' event definition",
           "pattern": "^[a-z0-9](-?[a-z0-9])*$"
         },
-        "consumeEventRef": {
-          "type": "string",
-          "description": "Reference to the unique name of a 'consumed' event definition",
-          "pattern": "^[a-z0-9](-?[a-z0-9])*$"
-        },
-        "consumeEventTimeout": {
-          "type": "string",
-          "description": "Maximum amount of time (ISO 8601 format) to wait for the result event. If not defined it should default to the actionExecutionTimeout"
-        },
         "data": {
           "type": [
             "string",
             "object"
           ],
-          "description": "If string type, an expression which selects parts of the states data output to become the data (payload) of the event referenced by 'produceEventRef'. If object type, a custom object to become the data (payload) of the event referenced by 'produceEventRef'."
+          "description": "If string type, an expression which selects parts of the states data output to become the data (payload) of the event referenced by 'publish'. If object type, a custom object to become the data (payload) of the event referenced by 'publish'."
         },
         "contextAttributes": {
           "type": "object",
@@ -556,7 +557,26 @@
       },
       "additionalProperties": false,
       "required": [
-        "produceEventRef"
+        "event", "data"
+      ]
+    },
+    "subscribe": {
+      "type": "object",
+      "description": "Subscribe to an event channel",
+      "properties": {
+        "event": {
+          "type": "string",
+          "description": "Reference to the unique name of a 'subscribed' event definition",
+          "pattern": "^[a-z0-9](-?[a-z0-9])*$"
+        },
+        "timeout": {
+          "type": "string",
+          "description": "Maximum amount of time (ISO 8601 format) to wait for the result event. If not defined it should default to the actionExecutionTimeout"
+        }
+       },
+      "additionalProperties": false,
+      "required": [
+        "event"
       ]
     },
     "subflowref": {
@@ -1690,7 +1710,7 @@
           "name",
           "type",
           "action",
-          "eventRef"
+          ""
         ]
       },
       "else": {