feat: add 'Metadata' field in schema.TaskDefinition

Author: Lack30

gateway_event_based.go

@@ -18,7 +18,6 @@ package bpmn
 
 import (
 	"context"
-	"fmt"
 	"sync"
 	"sync/atomic"
 
@@ -61,7 +60,7 @@ func (gw *eventBasedGateway) run(ctx context.Context, sender tracing.ISenderHand
 						terminationChannels[*idPtr] = make(chan bool)
 					} else {
 						err := errors.NotFoundError{
-							Expected: fmt.Sprintf("id for %#v", sequenceFlow),
+							Expected: sequenceFlow,
 						}
 						gw.tracer.Send(ErrorTrace{Error: err})
 					}

gateway_exclusive.go

@@ -148,7 +148,6 @@ func (gw *exclusiveGateway) run(ctx context.Context, sender tracing.ISenderHandl
 			case nextActionMessage:
 				if _, ok := gw.probing[m.flow.Id()]; ok {
 					gw.probing[m.flow.Id()] = &m.response
-					// and now we wait until the probe has returned
 				} else {
 					gw.probing[m.flow.Id()] = nil
 					m.response <- probeAction{

process_set.go

@@ -44,7 +44,7 @@ type ProcessSet struct {
 
 	definitions *schema.Definitions
 
-	sourceMessageFlows map[string]*schema.MessageFlow
+	messageFlows map[string]*schema.MessageFlow
 
 	cmu     sync.RWMutex
 	catchCh map[string]chan struct{}
@@ -94,25 +94,23 @@ func NewProcessSet(executeProcesses, waitingProcesses []*schema.Process, definit
 		executes = append(executes, process)
 	}
 
-	sourceMessageFlows := make(map[string]*schema.MessageFlow)
+	messageFlows := make(map[string]*schema.MessageFlow)
 	for _, collaboration := range *definitions.Collaborations() {
 		for _, msg := range *collaboration.MessageFlows() {
-			sourceMessageFlows[string(msg.SourceRefField)] = &msg
+			messageFlows[string(msg.SourceRefField)] = &msg
 		}
 	}
 
 	ps := &ProcessSet{
-		Options:            options,
-		sourceOptions:      opts,
-		executes:           executes,
-		waitings:           waitingProcesses,
-		definitions:        definitions,
-		sourceMessageFlows: sourceMessageFlows,
-
-		catchCh: make(map[string]chan struct{}),
-
-		mch:  make(chan imessage, len(executes)+1),
-		done: make(chan struct{}, 1),
+		Options:       options,
+		sourceOptions: opts,
+		executes:      executes,
+		waitings:      waitingProcesses,
+		definitions:   definitions,
+		messageFlows:  messageFlows,
+		catchCh:       make(map[string]chan struct{}),
+		mch:           make(chan imessage, len(executes)+1),
+		done:          make(chan struct{}, 1),
 	}
 
 	return ps, nil
@@ -160,7 +158,7 @@ func (ps *ProcessSet) run(ctx context.Context) {
 		case ch := <-ps.mch:
 			switch msg := ch.(type) {
 			case throwMessage:
-				sourceRef, ok := ps.sourceMessageFlows[msg.Id]
+				sourceRef, ok := ps.messageFlows[msg.Id]
 				if ok {
 					startFlowNode, waitingProcess, found := ps.resolveWaitingProcessAndEvent(string(sourceRef.TargetRefField))
 					if found {
@@ -184,9 +182,9 @@ func (ps *ProcessSet) run(ctx context.Context) {
 						ps.wg.Add(1)
 						go ps.tracerProcess(ctx, process, &ps.wg)
 					}
-					trigger, found := ps.triggerCatch(string(sourceRef.TargetRefField))
+					cancel, found := ps.triggerCatch(string(sourceRef.TargetRefField))
 					if found {
-						trigger()
+						cancel()
 					}
 				}
 			}

schema/schema.go

@@ -31,7 +31,7 @@ const (
 	BpmnDINS = "bpmndi:"
 	DINS     = "di:"
 	DCNS     = "dc:"
-	OliveNS  = "olive:"
+	OliveNS  = "olive:" // olive consumer namespace
 )
 
 var mapping = map[string]string{
@@ -356,6 +356,39 @@ type TaskDefinition struct {
 	Timeout string `xml:"timeout,attr"`
 	Retries int32  `xml:"retries,attr"`
 	Target  string `xml:"target,attr"`
+	// more task definition data (json format)
+	Metadata string `xml:"metadata,attr"`
+}
+
+func (t *TaskDefinition) SetMetadata(name string, value any) {
+	if t.Metadata == "" {
+		t.Metadata = "{}"
+	}
+	metadata := map[string]any{}
+	_ = json.Unmarshal([]byte(t.Metadata), &metadata)
+	metadata[name] = value
+	data, _ := json.Marshal(metadata)
+	t.Metadata = string(data)
+}
+
+func (t *TaskDefinition) GetMetadata(name string) (any, bool) {
+	metadata := t.GetMetadatas()
+	value, ok := metadata[name]
+	return value, ok
+}
+
+func (t *TaskDefinition) SetMetadatas(metadata map[string]any) {
+	data, _ := json.Marshal(metadata)
+	t.Metadata = string(data)
+}
+
+func (t *TaskDefinition) GetMetadatas() map[string]any {
+	if t.Metadata == "" {
+		return map[string]any{}
+	}
+	metadata := map[string]any{}
+	_ = json.Unmarshal([]byte(t.Metadata), &metadata)
+	return metadata
 }
 
 func (t *TaskDefinition) MarshalXML(e *xml.Encoder, start xml.StartElement) error {

schema/schema_parser_test.go

@@ -71,10 +71,10 @@ func TestParseSample(t *testing.T) {
 	if !assert.True(t, ok) {
 		return
 	}
-
-	_ = extension
-	t.Log(extension.TaskHeaderField.Header[0])
-	t.Log(extension.TaskDefinitionField.Type)
+	metadata := extension.TaskDefinitionField.GetMetadatas()
+	assert.Equal(t, map[string]any{"a": "b"}, metadata)
+	assert.Equal(t, "application/json", extension.TaskHeaderField.Header[0].Value)
+	assert.Equal(t, "service", extension.TaskDefinitionField.Type)
 }
 
 func TestParseSampleNs(t *testing.T) {

schema/testdata/sample.bpmn

@@ -30,9 +30,9 @@
       <bpmn:incoming>x3</bpmn:incoming>
       <bpmn:outgoing>x5</bpmn:outgoing>
       <bpmn:extensionElements>
-        <olive:taskDefinition type="service"/>
+        <olive:taskDefinition type="service" metadata="{&#34;a&#34;: &#34;b&#34;}"/>
         <olive:taskHeaders>
-          <olive:header name="contentType" value="aplication/json"/>
+          <olive:header name="contentType" value="application/json"/>
         </olive:taskHeaders>
         <olive:properties>
           <olive:property name="a" value="1" type="integer"/>