activity

activity.go

@@ -345,6 +345,7 @@ type TaskTrace struct {
 	dataObjects map[string]any
 	headers     map[string]any
 	properties  map[string]any
+	forward     chan DoResponse
 	response    chan DoResponse
 	done        chan struct{}
 }
@@ -356,6 +357,7 @@ func newTaskTrace() *TaskTrace {
 		dataObjects: make(map[string]any),
 		headers:     make(map[string]any),
 		properties:  make(map[string]any),
+		forward:     make(chan DoResponse, 1),
 		response:    make(chan DoResponse, 1),
 		done:        make(chan struct{}, 1),
 	}
@@ -396,9 +398,7 @@ func (t *TaskTrace) Do(options ...DoOption) {
 	}
 
 	response := newDoOption(options...)
-
-	t.response <- *response
-	close(t.done)
+	t.forward <- *response
 }
 
 func (t *TaskTrace) process() {
@@ -407,22 +407,27 @@ func (t *TaskTrace) process() {
 		duration = time.Second
 	}
 
-	for {
-		select {
-		case <-t.done:
-			return
-		case <-t.ctx.Done():
-			rsp := newDoOption(WithErr(t.ctx.Err()))
-			t.response <- *rsp
-		case <-time.After(duration):
-			var tid string
-			if v, ok := t.activity.Element().Id(); ok {
-				tid = *v
-			}
-
-			rsp := newDoOption(WithErr(errors.TaskExecError{Id: tid, Reason: "timed out"}))
-			t.response <- *rsp
+	select {
+	case <-t.done:
+	case <-t.ctx.Done():
+		rsp := newDoOption(WithErr(t.ctx.Err()))
+		t.response <- *rsp
+	case <-time.After(duration):
+		var tid string
+		if v, ok := t.activity.Element().Id(); ok {
+			tid = *v
 		}
+
+		rsp := newDoOption(WithErr(errors.TaskExecError{Id: tid, Reason: "timed out"}))
+		t.response <- *rsp
+	case rsp := <-t.forward:
+		t.response <- rsp
+	}
+
+	select {
+	case <-t.done:
+	default:
+		close(t.done)
 	}
 }
 

examples/basic/main.go

@@ -23,6 +23,7 @@ import (
 	"encoding/xml"
 	"io"
 	"log"
+	"time"
 
 	"github.com/olive-io/bpmn/schema"
 	"github.com/olive-io/bpmn/v2"
@@ -146,8 +147,8 @@ func main() {
 		switch tr := trace.(type) {
 		case *bpmn.TaskTrace:
 			log.Printf("%#v\n", trace)
-			_ = tr
-			//tr.Do()
+			time.Sleep(time.Second * 1)
+			tr.Do()
 		case bpmn.ErrorTrace:
 			log.Fatalf("%#v", trace)
 		default:

examples/basic/task.bpmn

@@ -10,7 +10,7 @@
         <olive:taskDefinition type="grpc" retries="2" />
         <olive:taskHeaders>
           <olive:header name="ov:content_type" value="application/grpc+json" />
-          <olive:header name="timeout" value="1s" />
+          <olive:header name="timeout" value="3s" />
           <olive:header name="ov:handler" value="rpc" />
           <olive:header name="ov:method" value="/gatewaypb.TestService/Hello" />
           <olive:header name="ov:url" value="/gatewaypb.TestService/Hello" />

flow.go

@@ -38,13 +38,9 @@ type Snapshot struct {
 	sequenceFlow *SequenceFlow
 }
 
-func (s *Snapshot) Id() id.Id {
-	return s.flowId
-}
+func (s *Snapshot) Id() id.Id { return s.flowId }
 
-func (s *Snapshot) SequenceFlow() *SequenceFlow {
-	return s.sequenceFlow
-}
+func (s *Snapshot) SequenceFlow() *SequenceFlow { return s.sequenceFlow }
 
 // Flow Represents a flow
 type Flow struct {

flow_wiring.go

@@ -90,15 +90,15 @@ func NewWiring(
 	if err != nil {
 		return
 	}
-	var ownId string
-	if ownIdPtr, present := flowNode.Id(); !present {
+	ownIdPtr, present := flowNode.Id()
+	if !present {
 		err = errors.NotFoundError{
 			Expected: fmt.Sprintf("flow node %#v to have an ID", flowNode),
 		}
 		return
-	} else {
-		ownId = *ownIdPtr
 	}
+
+	ownId := *ownIdPtr
 	node = &Wiring{
 		ProcessInstanceId:              processInstanceId,
 		FlowNodeId:                     ownId,
@@ -127,7 +127,6 @@ func (wiring *Wiring) CloneFor(flowNode *schema.FlowNode) (result *Wiring, err e
 	if err != nil {
 		return
 	}
-	var ownId string
 
 	ownIdPtr, present := flowNode.Id()
 	if !present {
@@ -137,7 +136,7 @@ func (wiring *Wiring) CloneFor(flowNode *schema.FlowNode) (result *Wiring, err e
 		return
 	}
 
-	ownId = *ownIdPtr
+	ownId := *ownIdPtr
 	result = &Wiring{
 		ProcessInstanceId:              wiring.ProcessInstanceId,
 		FlowNodeId:                     ownId,

subprocess.go

@@ -555,7 +555,7 @@ func (p *SubProcess) runner(ctx context.Context, out tracing.ITracer) {
 						}
 
 						trace = tracing.Unwrap(trace)
-						switch trace := trace.(type) {
+						switch tr := trace.(type) {
 						case CeaseFlowTrace:
 							out.Trace(ProcessLandMarkTrace{Node: p.element})
 							break loop
@@ -564,7 +564,7 @@ func (p *SubProcess) runner(ctx context.Context, out tracing.ITracer) {
 						case TerminationTrace:
 							// ignore end event of sub process
 						default:
-							out.Trace(trace)
+							out.Trace(tr)
 						}
 					}
 					p.tracer.Unsubscribe(traces)

task.go

@@ -90,7 +90,6 @@ func (task *Task) runner(ctx context.Context) {
 						task.Tracer.Trace(CancellationFlowNodeTrace{Node: task.element})
 						return
 					case <-at.out():
-
 					}
 
 					m.response <- action