func (wth *workflowTaskHandlerImpl) ProcessWorkflowTask(
	workflowTask *workflowTask,
	heartbeatFunc decisionHeartbeatFunc,
) (completeRequest interface{}, errRet error) {
	if workflowTask == nil || workflowTask.task == nil {
		return nil, errors.New("nil workflow task provided")
	}
	task := workflowTask.task
	if task.History == nil || len(task.History.Events) == 0 {
		task.History = &s.History{
			Events: []*s.HistoryEvent{},
		}
	}
	if task.Query == nil && len(task.History.Events) == 0 {
		return nil, errors.New("nil or empty history")
	}

	if task.Query != nil && len(task.Queries) != 0 {
		return nil, errors.New("invalid query decision task")
	}

	runID := task.WorkflowExecution.GetRunId()
	workflowID := task.WorkflowExecution.GetWorkflowId()
	traceLog(func() {
		wth.logger.Debug("Processing new workflow task.",
			zap.String(tagWorkflowType, task.WorkflowType.GetName()),
			zap.String(tagWorkflowID, workflowID),
			zap.String(tagRunID, runID),
			zap.Int64("PreviousStartedEventId", task.GetPreviousStartedEventId()))
	})

	workflowContext, err := wth.getOrCreateWorkflowContext(task, workflowTask.historyIterator)
	if err != nil {
		return nil, err
	}

	defer func() {
		workflowContext.Unlock(errRet)
	}()

	var response interface{}
process_Workflow_Loop:
	for {
		startTime := time.Now()
		response, err = workflowContext.ProcessWorkflowTask(workflowTask)
		if err == nil && response == nil {
		wait_LocalActivity_Loop:
			for {
				deadlineToTrigger := time.Duration(float32(ratioToForceCompleteDecisionTaskComplete) * float32(workflowContext.GetDecisionTimeout()))
				delayDuration := startTime.Add(deadlineToTrigger).Sub(time.Now())
				select {
				case <-time.After(delayDuration):
					// force complete, call the decision heartbeat function
					workflowTask, err = heartbeatFunc(
						workflowContext.CompleteDecisionTask(workflowTask, false),
						startTime,
					)
					if err != nil {
						return nil, &decisionHeartbeatError{Message: fmt.Sprintf("error sending decision heartbeat %v", err)}
					}
					if workflowTask == nil {
						return nil, nil
					}
					continue process_Workflow_Loop

				case lar := <-workflowTask.laResultCh:
					// local activity result ready
					response, err = workflowContext.ProcessLocalActivityResult(workflowTask, lar)
					if err == nil && response == nil {
						// decision task is not done yet, still waiting for more local activities
						continue wait_LocalActivity_Loop
					}
					break process_Workflow_Loop
				}
			}
		} else {
			break process_Workflow_Loop
		}
	}
	return response, err
}
// ProcessWorkflowTask processes all the events of the workflow task.
ProcessWorkflowTask processes all the events of the workflow task.
// ProcessWorkflowTask processes all the events of the workflow task .
func (wth *workflowTaskHandlerImpl) ProcessWorkflowTask(
	workflowTask *workflowTask,
	heartbeatFunc decisionHeartbeatFunc,
) (completeRequest interface{}, errRet error) {
	if workflowTask == nil || workflowTask.task == nil {
		return nil, errors.New("nil workflow task provided")
	}
	task := workflowTask.task
	if task.History == nil || len(task.History.Events) == 0 {
		task.History = &s.History{
			Events: []*s.HistoryEvent{},
		}
	}
	if task.Query == nil && len(task.History.Events) == 0 {
		return nil, errors.New("nil or empty history")
	}
	if task.Query != nil && len(task.Queries) != 0 {
		return nil, errors.New("invalid query decision task")
	}
	runID := task.WorkflowExecution.GetRunId()
	workflowID := task.WorkflowExecution.GetWorkflowId()
	traceLog(func() {
		wth.logger.Debug("Processing new workflow task.",
			zap.String(tagWorkflowType, task.WorkflowType.GetName()),
			zap.String(tagWorkflowID, workflowID),
			zap.String(tagRunID, runID),
			zap.Int64("PreviousStartedEventId", task.GetPreviousStartedEventId()))
	})
	workflowContext, err := wth.getOrCreateWorkflowContext(task, workflowTask.historyIterator)
	if err != nil {
		return nil, err
	}
	defer func() {
		workflowContext.Unlock(errRet)
	}()
	var response interface{}
process_Workflow_Loop:
	for {
		startTime := time.Now()
		response, err = workflowContext.ProcessWorkflowTask(workflowTask)
		if err == nil && response == nil {
		wait_LocalActivity_Loop:
			for {
				deadlineToTrigger := time.Duration(float32(ratioToForceCompleteDecisionTaskComplete) * float32(workflowContext.GetDecisionTimeout()))
				delayDuration := startTime.Add(deadlineToTrigger).Sub(time.Now())
				select {
				case <-time.After(delayDuration):
					workflowTask, err = heartbeatFunc(
						workflowContext.CompleteDecisionTask(workflowTask, false),
						startTime,
					)
					if err != nil {
						return nil, &decisionHeartbeatError{Message: fmt.Sprintf("error sending decision heartbeat %v", err)}
					}
					if workflowTask == nil {
						return nil, nil
					}
					continue process_Workflow_Loop
				case lar := <-workflowTask.laResultCh:
					response, err = workflowContext.ProcessLocalActivityResult(workflowTask, lar)
					if err == nil && response == nil {
						continue wait_LocalActivity_Loop
					}
					break process_Workflow_Loop
				}
			}
		} else {
			break process_Workflow_Loop
		}
	}
	return response, err
}
func ( wth * workflowTaskHandlerImpl ) ProcessWorkflowTask ( workflowTask * workflowTask , heartbeatFunc decisionHeartbeatFunc , ) ( completeRequest interface { } , errRet error ) { if workflowTask == nil || workflowTask . task == nil { return nil , errors . New ( " " ) } task := workflowTask . task if task . History == nil || len ( task . History . Events ) == 0 { task . History = & s . History { Events : [ ] * s . HistoryEvent { } , } } if task . Query == nil && len ( task . History . Events ) == 0 { return nil , errors . New ( " " ) } if task . Query != nil && len ( task . Queries ) != 0 { return nil , errors . New ( " " ) } runID := task . WorkflowExecution . GetRunId ( ) workflowID := task . WorkflowExecution . GetWorkflowId ( ) traceLog ( func ( ) { wth . logger . Debug ( " " , zap . String ( tagWorkflowType , task . WorkflowType . GetName ( ) ) , zap . String ( tagWorkflowID , workflowID ) , zap . String ( tagRunID , runID ) , zap . Int64 ( " " , task . GetPreviousStartedEventId ( ) ) ) } ) workflowContext , err := wth . getOrCreateWorkflowContext ( task , workflowTask . historyIterator ) if err != nil { return nil , err } defer func ( ) { workflowContext . Unlock ( errRet ) } ( ) var response interface { } process_Workflow_Loop : for { startTime := time . Now ( ) response , err = workflowContext . ProcessWorkflowTask ( workflowTask ) if err == nil && response == nil { wait_LocalActivity_Loop : for { deadlineToTrigger := time . Duration ( float32 ( ratioToForceCompleteDecisionTaskComplete ) * float32 ( workflowContext . GetDecisionTimeout ( ) ) ) delayDuration := startTime . Add ( deadlineToTrigger ) . Sub ( time . Now ( ) ) select { case <- time . After ( delayDuration ) : workflowTask , err = heartbeatFunc ( workflowContext . CompleteDecisionTask ( workflowTask , false ) , startTime , ) if err != nil { return nil , & decisionHeartbeatError { Message : fmt . Sprintf ( " " , err ) } } if workflowTask == nil { return nil , nil } continue process_Workflow_Loop case lar := <- workflowTask . laResultCh : response , err = workflowContext . ProcessLocalActivityResult ( workflowTask , lar ) if err == nil && response == nil { continue wait_LocalActivity_Loop } break process_Workflow_Loop } } } else { break process_Workflow_Loop } } return response , err }
ProcessWorkflowTask processes all the events of the workflow task.
ProcessWorkflowTask processes all the events of the workflow task .
// force complete, call the decision heartbeat function // local activity result ready // decision task is not done yet, still waiting for more local activities